The project I am working on is split up into several modules. Previously,
each module had its own ORM classes.
However, due to several bugs arising from forgetting to update each
module's ORM classes in lock step when adding new functionality, we have
decided it would be best to extract the ORM classes which interact with our
DB into their own module and make that available to the other modules via
pip.
One of these modules, which monitors for changes in an upstream DB and
applies them to ours, has some methods which are not present in the other
modules and which cannot be easily extracted out due to its dependencies on
upstream DB functionality.
As such, in this module, we must subclass the ORM models which interact
with our DB:
models.apimodels.db.person.py:
#...
@dataclass(init=False, eq=True, unsafe_hash=True)
class Person(Base):
__tablename__ = "person"
id: Mapped[int] = mapped_column(primary_key=True)
first_name: Mapped[str]
last_name: Mapped[str]
email_address: Mapped[str]
office_address: Mapped[str]
office_phone_number: Mapped[str]
# ...
etl.models.api_db.api_person.py:
#...
from apimodels.db.person import Person as PersonBase
# ...
class API_Person(PersonBase):
__tablename__ = "person"
__table_args__ = {"keep_existing": True}
def get_pvi(self):
# ...
def get_historical_pvis(self) -> list[str]:
# ...
def __eq__(self):
# ...
def __hash__(self):
# ...
@staticmethod
def from_upstream_hub_person(
uh_person: Optional[UH_Person],
) -> Optional["API_Person"]:
# ...
Of note is that this subclass does not add any new attributes or modify
existing ones, it merely adds some helper methods related to identifying
primary key changes in the upstream DB. (This is also why we override the
eq and hash methods provided by dataclasses - incoming changesets have to
be matched against existing records, even when primary keys change
upstream.)
This is effectively single-table inheritance, but it is not a good fit for
polymorphic_identity since it is not a distinct class, merely adding
module-specific helper methods, and if I am reading the documentation
correctly, using polymorphic_identity would mean any records touched by the
API_Person subclass (which is all of them) would no longer be usable by
other modules, which do not extend any of the models.
>From what I have read, it seems like the keep_existing param should be of
use here, but neither it nor extend_existing set to True help with the
errors I am seeing when attempting to interact with this subclass in any
way:
sqlalchemy.orm.exc.FlushError: Attempting to flush an item of type <class
'model.api_db.api_person.API_Person'> as a member of collection
"Identifier.person". Expected an object of type <class
'apimodels.db.person.Person'> or a polymorphic subclass of this type. If
<class 'model.api_db.api_person.API_Person'> is a subclass of <class
'apimodels.db.person.Person'>, configure mapper "Mapper[Person(person)]" to
load this subtype polymorphically, or set enable_typechecks=False to allow
any subtype to be accepted for flush.
I did try setting enable_typechecks to False, but this results in a
different error when attempting to use getattr on the subclass:
AttributeError: 'Person' object has no attribute 'meta'
Is there a better way of doing this?
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/bff6d34f-ea35-4aba-ba94-5ae1f29154fan%40googlegroups.com.
