Ошибка в реализации однонаправленной связи один-к-одному в SQLAlchemy

Не могли бы вы мне помочь разобраться в моей проблеме? Используя SQLAlchemy делаю 2 класса Server и State. Соответственно создаются 2 таблицы servers и state.

Таблица state будет хранить некоторое состояние окна приложения. Такие вещи как сервер (по-умолчанию).

Серверу не нужно знать установлен он по-умолчанию или нет. Следовательно здесь мне нужна однонаправленная связь один-к-одному.

Теперь перейдём к коду в модуле (scheme.py):

from sqlalchemy import ForeignKey
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


class Server(Base):
    __tablename__ = "servers"
    id: Mapped[int] = mapped_column(primary_key=True)
    domain: Mapped[str | None]
    ip: Mapped[str | None]
    port: Mapped[int]
    alias: Mapped[str | None]

    favorite_id: Mapped[int | None] = mapped_column(ForeignKey("state.id"))


class State(Base):
    __tablename__ = "state"
    id: Mapped[int] = mapped_column(primary_key=True)

    favorite_server: Mapped[Server | None] = relationship(uselist=False)

Если я правильно понимаю как это работает, то связь делается как один-ко-многим, но на множественной стороне указывается обычный объект, а не коллекция и в relationship устанавливается в False атрибут uselist.

Дальше я пробую следующий тестовый код:

import scheme
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session


# Сделаем базу SQLite3 в памяти
engine = create_engine("sqlite://", echo=True)

# Создадим схему базы данных
scheme.Base.metadata.create_all(engine)

# Внесём пару записей в таблицу servers
with Session(engine) as session:
    one_server = Server(domain="www.example.com", port=8874, alias="example")
    two_server = Server(ip="247.234.112.90", port=8874)
    session.add_all([one_server, two_server])
    session.commit()

# До этого момента всё работает хорошо. Логи базы данных опустил.

# Теперь попробуем создать новую запись в таблицу state и установить сервер по умолчанию.
with Session(engine) as db:
    stmt_1 = select(Server).where(Server.id == 2)
    serv = db.scalar(stmt_1)
    state = scheme.State()
    state.favorite_server = serv
    db.add(state)
    db.commit()

Этот код не работает. Он выдаёт следующую ошибку:

raise exc.FlushError(
sqlalchemy.orm.exc.FlushError: Attempting to flush an item of type <class 'modloader.client.data.scheme.Server'> as a member of collection "State.favorite_server". Expected an object of type <class 'modloader.client.data.scheme.Server'> or a polymorphic subclass of this type.

Судя по всему он получил то, что ожидал, но остался недоволен. Что я делаю не так?


Ответы (1 шт):

Автор решения: Dezox

Ошибка крылась в select. Когда был сделан общий импорт модуля scheme, классы стали доступны под именами scheme.State и scheme.Server.

Следовательно в select нужно было указать scheme.Server.

Изменение этой строки:

stmt_1 = select(Server).where(Server.id == 2)

на эту:

stmt_1 = select(scheme.Server).where(scheme.Server.id == 2)

дало нужный результат.

→ Ссылка