Dokumentasi/database SQL

Gunakan SQLite dengan Nexy

SQLite adalah database yang digunakan proyek Nexy baru ketika DATABASE_URL tidak disetel: satu file, tanpa server, dan driver Python sendiri, melalui SQLModel, SQLAlchemy atau Tortoise-ORM.

Ketika SQLite cocok

• Pengembangan, pengujian, dan demo: tidak ada yang perlu diinstal atau dimulai.
• Sebuah situs dengan satu server dan menulis sederhana: blog, alat internal, toko kecil.
• Data berat yang dapat dibaca dikirimkan bersama aplikasi.

Pilih PostgreSQL ketika beberapa server menulis sekaligus, atau ketika Anda memerlukan tipe kaya dan pencarian teks lengkap dalam skala besar.

String koneksi

DATABASE_URL=sqlite:///dev.db

Mesinnya

nx new menulis file ini ketika Anda memilih SQLModel. Satu baris yang ditambahkan untuk SQLite adalah check_same_thread: halaman dan pengendali dapat menggunakan mesin dari thread yang berbeda.

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},)

Kunci asing dan WAL

SQLite tidak menerapkan kunci asing kecuali diminta, per koneksi. Mode WAL memungkinkan pembaca melanjutkan sementara seorang penulis menulis. Keduanya adalah PRAGMA yang dijalankan pada setiap koneksi baru, dipasang di sebelah mesin:

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()

Seorang model

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

Sesi per permintaan

Ketergantungan membuka sesi untuk permintaan dan menutupnya setelah jawabannya:

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

Membaca dan menulis dari seorang pawang

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)

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)

Membaca dari satu halaman

Halaman .nexy menjalankan frontmatternya di server, sehingga dapat memanggil kode yang sama. Pertahankan kueri dalam suatu fungsi: hanya nama yang ditetapkan di tingkat atas materi depan yang mencapai templat.

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

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

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

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")

Buat tabel dengan migrasi

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

--init menulis alembic/env.py, yang mengimpor model yang dibandingkan dengan database. Impor setiap modul model baru di sana:

import src.models.note_model  # noqa: F401

SQLite tidak dapat ALTER sebagian besar kolom terpasang. Aktifkan mode batch Alembic, yang menyalin tabel, di kedua panggilan context.configure(...) dari alembic/env.py:

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

Asinkron dengan aiosqlite

uv add aiosqlite

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

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}

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}

Pengujian pada database dalam memori

StaticPool menyimpan satu koneksi, sehingga database dalam memori tetap aktif selama keseluruhan pengujian. Ketergantungan handler ditukar dengan yang menggunakannya:

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"

Dalam produksi

• Simpan berkas pada disk persisten. Sistem berkas milik kontainer hilang saat penerapan ulang: pasang volume dan arahkan DATABASE_URL ke dalamnya dengan jalur absolut.
• Satu server, satu file. SQLite memiliki satu penulis pada satu waktu. Beberapa nx start pekerja pada mesin yang sama baik-baik saja dengan WAL; beberapa mesin tidak.
• Cadangkan saat dijalankan dengan sqlite3 app.db ".backup backup.db", bukan dengan salinan file.

Langkah selanjutnya

• Basis Data
• Gunakan PostgreSQL dengan Nexy
• Pengujian

Salin halaman sebagai Markdown Sunting di GitHub