SQLite сохранила ссылку на несуществующую запись: включаем внешние ключи

У задачи есть project_id и объявленный ForeignKey. API создаёт задачу с номером проекта, которого нет в таблице, и отвечает 201. После перехода на другую базу тот же запрос внезапно начинает падать. Причина часто в том, что тестовая SQLite не проверяла внешние ключи, хотя описание схемы выглядело правильным.

Описание связи и её исполнение

ForeignKey в SQLAlchemy создаёт ограничение в схеме, но SQLite требует включить его исполнение для подключения. Настройка PRAGMA foreign_keys=ON действует на конкретное соединение. Один ручной вызов после запуска приложения не гарантирует, что следующая Session получит подключение с тем же режимом. При использовании пула это особенно легко пропустить.

Поэтому настройку удобно выполнять на событии connect у Engine. Обработчик вызывается при создании DBAPI-подключения, до его использования ORM. Зарегистрировать событие нужно до create_all и любых запросов: пул может успеть открыть соединения раньше. Уже созданные подключения не получают настройку задним числом только потому, что появился новый обработчик.

В примере обработчик учитывает наличие атрибута autocommit у sqlite3.Connection. Для выполнения PRAGMA вне транзакции он временно включает этот режим и восстанавливает исходное значение. На поддерживаемых версиях Python без этого атрибута выполняется обычный вызов до начала работы с соединением. Так код подходит и окружению Python 3.12 с актуальным драйвером.

Проверим ограничение через маршрут

Нужны FastAPI, HTTPX и SQLAlchemy 2. Для локального теста используется одна SQLite в памяти через StaticPool. В API намеренно нет предварительной проверки существования проекта: отказ даёт именно внешний ключ. В учебной таблице у запроса есть единственная ожидаемая причина IntegrityError, поэтому она переводится в 409.

В рабочей модели с несколькими ограничениями нельзя превращать любую IntegrityError в «проект отсутствует». Различайте ожидаемые нарушения по данным драйвера или выполняйте дополнительную диагностику. Полезный ответ клиенту должен соответствовать реальному конфликту, а неизвестная ошибка требует отдельного расследования.

from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel
from sqlalchemy import create_engine, event, ForeignKey, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.exc import IntegrityError
from sqlalchemy.pool import StaticPool

engine = create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False})
@event.listens_for(engine, "connect")
def enable_fk(connection, record):
    has_autocommit = hasattr(connection, "autocommit")
    if has_autocommit:
        previous = connection.autocommit
        connection.autocommit = True
    cursor = connection.cursor()
    try:
        cursor.execute("PRAGMA foreign_keys=ON")
    finally:
        cursor.close()
        if has_autocommit:
            connection.autocommit = previous

class Base(DeclarativeBase):
    pass
class Project(Base):
    __tablename__ = "project"
    id: Mapped[int] = mapped_column(primary_key=True)
class Task(Base):
    __tablename__ = "task"
    id: Mapped[int] = mapped_column(primary_key=True)
    project_id: Mapped[int] = mapped_column(ForeignKey("project.id"))
    name: Mapped[str] = mapped_column(String)

Base.metadata.create_all(engine)
with Session(engine) as db:
    db.add(Project(id=1))
    db.commit()
class NewTask(BaseModel):
    project_id: int
    name: str
app = FastAPI()
@app.post("/tasks", status_code=201)
def create(data: NewTask):
    with Session(engine) as db:
        row = Task(**data.model_dump())
        db.add(row)
        try:
            db.commit()
        except IntegrityError:
            db.rollback()
            raise HTTPException(409, "unknown project")
        return {"id": row.id, "project_id": row.project_id}

with TestClient(app) as client:
    bad = client.post("/tasks", json={"project_id": 999, "name": "bad"})
    good = client.post("/tasks", json={"project_id": 1, "name": "good"})
    assert bad.status_code == 409 and good.status_code == 201
with Session(engine) as db:
    count = db.scalar(select(func.count()).select_from(Task))
    assert count == 1
print(bad.status_code, good.status_code, count)
engine.dispose()

Что останется за пределами ForeignKey

Ограничение доказывает существование проекта, но не право текущего пользователя создавать в нём задачи. Для многопользовательского API нужна отдельная проверка владельца или участия. Не используйте успешную вставку как авторизацию: база проверяет структуру ссылок, а не продуктовую роль пользователя.

Предварительный SELECT проекта полезен для понятного 404 и проверки прав. Но он не заменяет ограничение: между чтением и вставкой проект могут удалить. Поэтому оба уровня имеют смысл. Сначала приложение принимает решение о доступе, затем база сохраняет целостность при конкурентных изменениях. После отказа сохранения Session требует rollback, прежде чем её можно использовать снова.

Добавьте тест удаления родительской записи. Поведение зависит от ondelete и настроек ORM-каскада: одни приложения запрещают удаление, другие удаляют дочерние задачи. Само включение foreign_keys не выбирает это правило. Оно лишь заставляет SQLite исполнять объявленное ограничение. Явно решите, что пользователь ожидает увидеть после удаления проекта.

Если тесты используют отдельный Engine, подключите обработчик и там. Частая ошибка — включить PRAGMA в production-фабрике, а тестовое приложение создать другим путём. Тогда тест снова принимает невозможные данные. Общая фабрика Engine и отрицательный HTTP-сценарий удерживают договор ближе к реальному приложению.

При переходе на PostgreSQL сохраните тест несуществующей ссылки. Реализация настройки будет другой, но требование останется тем же. Проверяйте результат через новую Session и фактическое число строк: объект Python мог получить id ещё до окончательного commit, поэтому существование объекта само по себе ничего не доказывает о сохранении.

Следующая практика

В бесплатной программе FastAPI есть Engine, Session, SQLite и CRUD через SQLAlchemy 2. Подключите эту основу и добавьте к тестам невозможную ссылку на проект: так настройка базы станет проверяемым поведением API. Перейдите к бесплатному курсу FastAPI.