الوثائققواعد بيانات SQL

استخدم SQLite مع Nexy

SQLite هي قاعدة البيانات التي يستخدمها مشروع Nexy الجديد عندما يتم إلغاء تعيين DATABASE_URL: ملف واحد، بدون خادم، وبرنامج التشغيل الخاص بـ Python، من خلال SQLModel، SQLAlchemy أو Tortoise-ORM.

عندما يناسب SQLite

التطوير والاختبارات والعروض التوضيحية: لا يوجد شيء للتثبيت أو البدء.
موقع بخادم واحد ومتواضع يكتب: مدونة، أداة داخلية، متجر صغير.
قراءة البيانات الثقيلة التي يتم شحنها مع التطبيق.

اختر PostgreSQL عندما تقوم عدة خوادم بالكتابة في وقت واحد، أو عندما تحتاج إلى أنواع منسقة والبحث عن النص الكامل على نطاق واسع.

سلسلة الاتصال

.env

DATABASE_URL=sqlite:///dev.db

عنوان URL	قاعدة البيانات
`sqlite:///dev.db`	`dev.db`، نسبة إلى المجلد الذي يعمل فيه `nx dev`
`sqlite:////var/data/app.db`	المسار المطلق: أربع خطوط مائلة
`sqlite://`	في الذاكرة، ويختفي عندما تتوقف العملية
`sqlite+aiosqlite:///dev.db`	نفس الملف، من خلال برنامج التشغيل غير المتزامن

المحرك

nx new يكتب هذا الملف عند اختيار SQLModel. السطر الوحيد الذي يجب إضافته لـ SQLite هو check_same_thread: قد تستخدم الصفحة والمعالج المحرك من سلاسل رسائل مختلفة.

src/configs/db.py

import os from dotenv import load_dotenvfrom sqlmodel import create_engine load_dotenv() engine = create_engine(    os.getenv("DATABASE_URL", "sqlite:///dev.db"),    connect_args={"check_same_thread": False},)

المفاتيح الخارجية وWAL

SQLite لا يفرض المفاتيح الخارجية إلا إذا طلب منها ذلك لكل اتصال. يتيح وضع WAL للقراء الاستمرار أثناء الكتابة. كلاهما PRAGMA يتم تشغيلهما عند كل اتصال جديد، ويتم إعدادهما بجوار المحرك:

src/configs/db.py

import os from dotenv import load_dotenvfrom sqlalchemy import eventfrom sqlmodel import create_engine load_dotenv() engine = create_engine(    os.getenv("DATABASE_URL", "sqlite:///dev.db"),    connect_args={"check_same_thread": False},)  @event.listens_for(engine, "connect")def set_pragmas(connection, _record) -> None:    cursor = connection.cursor()    cursor.execute("PRAGMA foreign_keys=ON")    cursor.execute("PRAGMA journal_mode=WAL")    cursor.close()

نموذج

src/models/note_model.py

from sqlmodel import Field, SQLModel  class Note(SQLModel, table=True):    id: int | None = Field(default=None, primary_key=True)    title: str = Field(max_length=200)    done: bool = False

جلسة لكل طلب

تفتح التبعية جلسة للطلب وتغلقها بعد الإجابة:

src/configs/session.py

from collections.abc import Iterator from sqlmodel import Session from src.configs.db import engine  def get_session() -> Iterator[Session]:    with Session(engine) as session:        yield session

القراءة والكتابة من المعالج

src/routes/api/notes/index.py

from pydantic import BaseModelfrom sqlmodel import Session, select from nexy.http import Depends, JSONResponsefrom src.configs.session import get_sessionfrom src.models.note_model import Note  class NewNote(BaseModel):    title: str  async def GET(session: Session = Depends(get_session)):    return session.exec(select(Note)).all()  async def POST(body: NewNote, session: Session = Depends(get_session)):    note = Note(title=body.title)    session.add(note)    session.commit()    session.refresh(note)    return JSONResponse(note.model_dump(), status_code=201)

src/apps/notes/notes_api_controller.py

from pydantic import BaseModelfrom sqlmodel import Session, select from nexy.decorators import Controllerfrom nexy.http import Depends, JSONResponsefrom src.configs.session import get_sessionfrom src.models.note_model import Note  class NewNote(BaseModel):    title: str  @Controller("/api/notes")class NotesApiController:    async def get(self, session: Session = Depends(get_session)):        return session.exec(select(Note)).all()     async def post(self, body: NewNote, session: Session = Depends(get_session)):        note = Note(title=body.title)        session.add(note)        session.commit()        session.refresh(note)        return JSONResponse(note.model_dump(), status_code=201)

قم بإدراج وحدة التحكم في controllers من الوحدة الجذرية.

إقرأ من إحدى الصفحات

تقوم صفحة .nexy بتشغيل الجزء الأمامي الخاص بها على الخادم، حتى تتمكن من استدعاء نفس الرمز. احتفظ بالاستعلام في دالة: فقط الأسماء المعينة في المستوى الأعلى من المادة الأمامية هي التي تصل إلى القالب.

src/services/notes.py

from sqlmodel import Session, select from src.configs.db import enginefrom src.models.note_model import Note  def open_notes() -> list[Note]:    with Session(engine) as session:        return list(session.exec(select(Note).where(Note.done == False)))  # noqa: E712

src/routes/notes.nexy

---from src.services.notes import open_notes notes = open_notes()---<ul>  {% for note in notes %}  <li>{{ note.title | e }}</li>  {% endfor %}</ul>

src/apps/notes/notes_view.nexy

---from src.services.notes import open_notes notes = open_notes()---<ul>  {% for note in notes %}  <li>{{ note.title | e }}</li>  {% endfor %}</ul>

src/apps/notes/notes_controller.py

from nexy.decorators import Controllerfrom nexy.hooks import useViews  @Controller("/notes")class NotesController:    async def get(self):        return useViews("src/apps/notes/notes_view.nexy")

إنشاء الجداول مع الهجرات

nx migrate --initnx migrate -m "create notes"nx migrate --upgrade

--init يكتب alembic/env.py، الذي يستورد النماذج التي يقارنها مع قاعدة البيانات. قم باستيراد كل وحدة نموذجية جديدة هناك:

الإنبك/env.py

import src.models.note_model  # noqa: F401

SQLite لا يمكن ALTER معظم الأعمدة في مكانها. قم بتشغيل الوضع الدفعي لـ Alembic، الذي ينسخ الجدول، في كلا استدعاءات context.configure(...) لـ alembic/env.py:

الإنبك/env.py

context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)

غير متزامن مع `aiosqlite`

uv add aiosqlite

src/configs/async_db.py

from sqlalchemy.ext.asyncio import create_async_enginefrom sqlmodel.ext.asyncio.session import AsyncSession async_engine = create_async_engine("sqlite+aiosqlite:///dev.db")  async def get_async_session():    async with AsyncSession(async_engine) as session:        yield session

src/routes/api/notes/count.py

from sqlmodel import func, selectfrom sqlmodel.ext.asyncio.session import AsyncSession from nexy.http import Dependsfrom src.configs.async_db import get_async_sessionfrom src.models.note_model import Note  async def GET(session: AsyncSession = Depends(get_async_session)):    count = (await session.exec(select(func.count()).select_from(Note))).one()    return {"count": count}

src/apps/notes/notes_count_api_controller.py

from sqlmodel import func, selectfrom sqlmodel.ext.asyncio.session import AsyncSession from nexy.decorators import Controllerfrom nexy.http import Dependsfrom src.configs.async_db import get_async_sessionfrom src.models.note_model import Note  @Controller("/api/notes/count")class NotesCountApiController:    async def get(self, session: AsyncSession = Depends(get_async_session)):        count = (await session.exec(select(func.count()).select_from(Note))).one()        return {"count": count}

قم بإدراج وحدة التحكم في controllers من الوحدة الجذرية.

اختبارات على قاعدة بيانات في الذاكرة

StaticPool يحتفظ باتصال واحد، وبالتالي فإن قاعدة البيانات الموجودة في الذاكرة تظل صالحة للاختبار بأكمله. يتم تبديل تبعية المعالج بأخرى تستخدمه:

الاختبارات/test_notes.py

from fastapi.testclient import TestClientfrom sqlalchemy.pool import StaticPoolfrom sqlmodel import Session, SQLModel, create_engine from nexy.app import appfrom src.configs.session import get_session test_engine = create_engine(    "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)SQLModel.metadata.create_all(test_engine)  def override_session():    with Session(test_engine) as session:        yield session  def test_a_note_is_created():    app.dependency_overrides[get_session] = override_session    client = TestClient(app)    client.post("/api/notes", json={"title": "Write the docs"})    assert client.get("/api/notes").json()[0]["title"] == "Write the docs"

في الإنتاج

احتفظ بالملف على قرص ثابت. يتم فقدان نظام الملفات الخاص بالحاوية عند إعادة النشر: قم بتحميل وحدة تخزين وأشر DATABASE_URL إليها بمسار مطلق.
خادم واحد، ملف واحد. SQLite لديه كاتب واحد في كل مرة. العديد من nx start العاملين على نفس الجهاز جيدون مع WAL؛ عدة آلات ليست كذلك.
قم بعمل نسخة احتياطية منه أثناء تشغيله مع sqlite3 app.db ".backup backup.db"، وليس مع نسخة ملف.

الخطوات التالية

حرِّر على GitHub