62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""Движок БД, PRAGMA для SQLite и зависимость сессии."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Iterator
|
|
|
|
from sqlalchemy import event
|
|
from sqlalchemy.engine import Engine
|
|
from sqlmodel import Session, create_engine
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def _ensure_sqlite_dir(url: str) -> None:
|
|
"""Создаёт каталог для файла SQLite (data/dev в деве, /data на томе в проде)."""
|
|
prefix = "sqlite:///"
|
|
if not url.startswith(prefix):
|
|
return
|
|
path = url[len(prefix):]
|
|
if path.startswith("/"): # абсолютный путь (sqlite:////...)
|
|
file_path = path
|
|
else:
|
|
file_path = path
|
|
directory = os.path.dirname(file_path)
|
|
if directory:
|
|
os.makedirs(directory, exist_ok=True)
|
|
|
|
|
|
_ensure_sqlite_dir(settings.database_url)
|
|
|
|
# check_same_thread=False — FastAPI работает в нескольких потоках.
|
|
_connect_args = (
|
|
{"check_same_thread": False}
|
|
if settings.database_url.startswith("sqlite")
|
|
else {}
|
|
)
|
|
|
|
engine = create_engine(
|
|
settings.database_url,
|
|
echo=False,
|
|
connect_args=_connect_args,
|
|
)
|
|
|
|
|
|
@event.listens_for(Engine, "connect")
|
|
def _set_sqlite_pragma(dbapi_connection, connection_record) -> None: # noqa: ANN001
|
|
"""Включает FK и настраивает WAL на каждом соединении SQLite."""
|
|
# Срабатывает для всех движков; PRAGMA применяем только к sqlite3.
|
|
if dbapi_connection.__class__.__module__.startswith("sqlite3"):
|
|
cursor = dbapi_connection.cursor()
|
|
cursor.execute("PRAGMA foreign_keys=ON")
|
|
cursor.execute("PRAGMA journal_mode=WAL")
|
|
cursor.execute("PRAGMA synchronous=NORMAL")
|
|
cursor.execute("PRAGMA busy_timeout=30000")
|
|
cursor.close()
|
|
|
|
|
|
def get_session() -> Iterator[Session]:
|
|
"""FastAPI-зависимость: сессия на запрос."""
|
|
with Session(engine) as session:
|
|
yield session
|