SQLAlchemy - связь приводит к сбою ограничения UNIQUE при слиянии ⇐ Python

[Anonymous]

Это может быть немного сложный вопрос, поэтому, пожалуйста, потерпите. Я нашел решение, которое работает, но думаю, что оно может быть ненадежным, и буду признателен за советы о том, как его улучшить.
Я создаю ORM для тестовой системы. . Базовая структура такова:

• Отдельные устройства представлены объектом DeviceInfo.
• Устройства классифицируются объектом DeviceFamily.
• Тесты выполняются на отдельных устройствах. Тесты создают объект TestResult.
• Тесты классифицируются по объекту TestType.

Вот классы SQLAlchemy и диаграмму для иллюстрации.

Код: Выделить всё

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Mapped, mapped_column, relationship, Session

class ProductFamily(Base):
__tablename__ = "product_family"

name: Mapped[str] = mapped_column(String(255), unique=True)
vendor_id: Mapped[int] = mapped_column(primary_key=True)
product_id: Mapped[int] = mapped_column(primary_key=True)
devices: Mapped[list["DeviceInfo"]] = relationship(back_populates="product_family")

class DeviceInfo(Base):
"""
Dataclass for device information. Compatible with sqlalchemy. API
"""

__tablename__ = "device"

serial: Mapped[str] = mapped_column(String(255), primary_key=True)
tests: Mapped[list["DeviceTestResult"]] = relationship(
back_populates="device",
)
tested_at: Mapped[datetime] = mapped_column(
insert_default=datetime.now(), default=None, nullable=True
)
product_family: Mapped[ProductFamily] = relationship(back_populates="devices")
product_family_vendor_id: Mapped[int] = mapped_column()
product_family_product_id: Mapped[int] = mapped_column()

# To handle composite foreign key
__table_args__ = (
ForeignKeyConstraint(
[product_family_vendor_id, product_family_product_id],
[ProductFamily.vendor_id, ProductFamily.product_id],
),
{},
)

class TestType(Base):
__tablename__ = "test_type"
name: Mapped[str] = mapped_column(primary_key=True)
# Product family has been removed as it raises an integrity error
# product_family: Mapped['ProductFamily'] = relationship(back_populates='test_types')
vendor_id: Mapped[int] = mapped_column(primary_key=True)
product_id: Mapped[int] = mapped_column(primary_key=True)
tests: Mapped[list['TestResult']] = relationship(back_populates='test_type')

class TestResult(Base):
__tablename__ = "device_test"

device: Mapped["DeviceInfo"] = relationship(back_populates="tests")
test_type: Mapped['DeviceTestType'] = relationship(back_populates='tests')
completed_at: Mapped[datetime] = mapped_column(
insert_default=datetime.now(), default=datetime.now()
)

idx: Mapped[int] = mapped_column(
primary_key=True, autoincrement=True, sort_order=-1
)
device_serial: Mapped[int] = mapped_column(
ForeignKey("device.serial"),
)
test_type_name: Mapped[str] = mapped_column()
test_type_vendor_id: Mapped[int] = mapped_column()
test_type_product_id: Mapped[int] = mapped_column()

__table_args__ = (
ForeignKeyConstraint(
[test_type_name, test_type_vendor_id, test_type_product_id],
[TestType.name, TestType.vendor_id, TestType.product_id],
),
{},
)

Результаты теста создаются классом DeviceTester. Я использую шаблон стратегии, позволяющий пользователям реализовать свои тесты с помощью этого класса.

Код: Выделить всё

class DeviceTester:
def __init__(self, device: DeviceInfo):
self.name = 'usb_test'
# Dynamically generate TestType
self.test_type = TestType(name=self.name, product_family=device.product_family)

def run_test(self):
# User implements test execution here

result = TestResult(test_type=self.test_type)
return result

Обратите внимание, что TestType генерируется внутри DeviceTester — это сделано для того, чтобы TestType не нужно было внедрять как зависимость. Я хочу, чтобы пользователь мог добавить новый тип теста, просто создав новый класс DeviceTester, вместо того, чтобы взаимодействовать с базой данных.
Проблема связана с связь между TestType и ProductFamily — поскольку TestType генерируется динамически, каждый DeviceTester имеет отдельный объект TestType. Я надеялся, что использование session.merge разрешит эти конфликты, поскольку отдельные объекты имеют одинаковые значения полей, но это все равно приводит к сбою ограничения UNIQUE.
Решение, которое у меня есть сейчас, было предложено, когда я в последний раз задавал аналогичный вопрос. Я накапливаю уникальные типы тестов, переназначаю TestResult.test_type и объединяю типы тестов. В моем тестировании простое слияние TestTypes каскадирует и добавляет все связанные объекты.

Код: Выделить всё

def add_test_results(session: Session, results: list[TestResult]):
test_types = {}

# Collect unique test types & reassign test result type
for result in results:
test_type = result.test_type
test_type_tuple = (test_type.name, result.device.product_family.product_id, result.device.product_family.vendor_id)

if test_type_tuple not in test_types:
test_types[test_type_tuple] = test_type
else:
result.test_type = test_types[test_type_tuple]

# Merge results
for test_type in test_types.values():
session.merge(test_type)

Некоторые другие решения, которые я рассмотрел:

• Удалите связь между ProductFamily и TestType, но все еще имеет Product_id иvendor_id в качестве первичных ключей в TestType
  
  Я пробовал это, и это работает - дубликаты TestTypes объединяются с session.merge, как и ожидалось.
• Для этого потребуется специальная функция запроса для поиска TestTypes, связанных с ProductFamily< /code>, а не просто просматривать объекты ORM.

[*]Объедините TestType и DeviceTester в один объект

• Это означает, что перед тестированием пользователь может запросить, существует ли тест, а затем обновить запись новыми результатами
• Однако, поскольку DeviceTester будет подклассом, пользователю придется беспокоиться о наследовании сопоставлений SQLAlchemy, что является для него головной болью.
• Я также думаю, что объекты данных & тестовые объекты не должны быть так тесно связаны, поскольку они представляют разные проблемы.

Есть ли какие-нибудь другие решения, которые могут сработать? Является ли моя конструкция базы данных фундаментально ошибочной? Любые советы по улучшению моего кода приветствуются.


Подробнее здесь: https://stackoverflow.com/questions/791 ... en-merging