Docs/bases de datos SQL

Utilice SQLite con Nexy

SQLite es la base de datos que utiliza un nuevo proyecto Nexy cuando DATABASE_URL no está configurado: un archivo, ningún servidor y el propio controlador de Python, a través de SQLModel, SQLAlchemy o Tortoise-ORM.

Cuando SQLite encaja

• Desarrollo, pruebas y demostraciones: nada que instalar o iniciar.
• Un sitio con un servidor y escrituras modestas: un blog, una herramienta interna, una pequeña tienda.
• Datos de gran cantidad de lectura enviados con la aplicación.

Elija PostgreSQL cuando varios servidores escriban a la vez o cuando necesite tipos enriquecidos y búsqueda de texto completo a escala.

La cadena de conexión

DATABASE_URL=sqlite:///dev.db

el motor

nx new escribe este archivo cuando eliges SQLModel. La única línea que se debe agregar para SQLite es check_same_thread: una página y un controlador pueden usar el motor de diferentes hilos.

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

Claves externas y WAL

SQLite no aplica claves externas a menos que se soliciten, por conexión. El modo WAL permite a los lectores continuar mientras un escritor escribe. Ambos se ejecutan PRAGMA en cada nueva conexión, configurados junto al motor:

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 modelo

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

Una sesión por solicitud

Una dependencia abre una sesión para la solicitud y la cierra después de la respuesta:

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

Leer y escribir desde un controlador.

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)

Leer de una página

Una página .nexy ejecuta su frontmatter en el servidor, por lo que puede llamar al mismo código. Mantener la consulta en una función: solo los nombres asignados en el nivel superior del frontmatter llegan a la plantilla.

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

Crear las tablas con migraciones.

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

--init escribe alembic/env.py, que importa los modelos que compara con la base de datos. Importe cada nuevo módulo de modelo allí:

import src.models.note_model  # noqa: F401

SQLite no puede ALTER la mayoría de las columnas en su lugar. Active el modo por lotes de Alembic, que copia la tabla, en ambas llamadas de context.configure(...) de alembic/env.py:

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

Asíncrono con 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}

Pruebas en una base de datos en memoria

StaticPool mantiene una conexión, por lo que la base de datos en memoria permanece viva durante toda la prueba. La dependencia del controlador se cambia por una que lo usa:

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 producción

• Mantenga el archivo en un disco persistente. El propio sistema de archivos de un contenedor se pierde al volver a implementarlo: monte un volumen y apúntelo DATABASE_URL con una ruta absoluta.
• Un servidor, un archivo. SQLite tiene un escritor a la vez. Varios nx start trabajadores en la misma máquina están bien con WAL; varias máquinas no lo son.
• Haga una copia de seguridad mientras se ejecuta con sqlite3 app.db ".backup backup.db", no con una copia del archivo.

Próximos pasos

• Bases de datos
• Utilice PostgreSQL con Nexy
• Pruebas

Copiar la página como Markdown Editar en GitHub