Docs/Bases de données SQL

Utiliser SQLite avec Nexy

SQLite est la base de données qu'un nouveau projet Nexy utilise lorsque DATABASE_URL n'est pas défini : un fichier, aucun serveur et le propre pilote de Python, via SQLModel, SQLAlchemy ou Tortoise-ORM.

Quand SQLite correspond

• Développement, tests et démos : rien à installer ni à démarrer.
• Un site avec un seul serveur et une écriture modeste : un blog, un outil interne, une petite boutique.
• Données lourdes en lecture livrées avec l'application.

Choisissez PostgreSQL lorsque plusieurs serveurs écrivent en même temps ou lorsque vous avez besoin de types riches et d'une recherche en texte intégral à grande échelle.

La chaîne de connexion

DATABASE_URL=sqlite:///dev.db

Le moteur

nx new écrit ce fichier lorsque vous sélectionnez SQLModel. La seule ligne à ajouter pour SQLite est check_same_thread : une page et un gestionnaire peuvent utiliser le moteur à partir de différents threads.

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

Clés étrangères et WAL

SQLite n'applique pas les clés étrangères sauf si cela est demandé, par connexion. Le mode WAL permet aux lecteurs de continuer pendant qu'un écrivain écrit. Les deux sont un PRAGMA exécuté à chaque nouvelle connexion, configuré à côté du moteur :

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

Un modèle

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

Une séance par demande

Une dépendance ouvre une session pour la requête et la ferme après la réponse :

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

Lire et écrire à partir d'un gestionnaire

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)

Lire à partir d'une page

Une page .nexy exécute son frontmatter sur le serveur, elle peut donc appeler le même code. Conservez la requête dans une fonction : seuls les noms attribués au niveau supérieur du frontmatter atteignent le modèle.

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

Créer les tables avec les migrations

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

--init écrit alembic/env.py, qui importe les modèles qu'il compare avec la base de données. Importez-y chaque nouveau module de modèle :

import src.models.note_model  # noqa: F401

SQLite ne peut pas ALTER la plupart des colonnes en place. Activez le mode batch de Alembic, qui copie la table, dans les deux appels context.configure(...) de alembic/env.py :

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

Asynchrone avec 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}

Tests sur une base de données en mémoire

StaticPool conserve une connexion, de sorte que la base de données en mémoire dure tout le test. La dépendance du gestionnaire est remplacée par celle qui l'utilise :

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"

En production

• Conservez le fichier sur un disque persistant. Le système de fichiers d'un conteneur est perdu lors du redéploiement : montez un volume et pointez DATABASE_URL dessus avec un chemin absolu.
• Un serveur, un fichier. SQLite n'a qu'un seul rédacteur à la fois. Plusieurs nx start travailleurs sur la même machine acceptent WAL ; plusieurs machines ne le sont pas.
• Sauvegardez-le pendant son exécution avec sqlite3 app.db ".backup backup.db", pas avec une copie de fichier.

Prochaines étapes

• Bases de données
• Utiliser PostgreSQL avec Nexy
• Tests

Copier la page en Markdown Modifier sur GitHub