How to implement overlapping joined table inheritance?

[benlau6]

Hi. I am implementing a supertype-subtype joined table inheritance as follows.

That is, a person might be a employee and student at the same time. At first, I followed SQLAlchemy1.4 Documentation - Mapping Class Inheritance Hierarchies. That's great, but I found that it is only for disjoint subclasses because the subclass table records are fully dependent on base class table's discriminator, i.e. I cannot insert the same person to two sub classes table.

I had googled to figure out that, I could use two discriminators, which are 'is_employee' and 'is_student' to solve the overlapping problem.

But then I found a note stating that

Note
Currently, only one discriminator column or SQL expression may be configured for the entire inheritance hierarchy, typically on the base- most class in the hierarchy. “Cascading” polymorphic discriminator expressions are not yet supported.

I was not sure what is 'Cascading', but I got that I can only use one discriminator column or SQL expression. So I tried PostgreSQL's ARRAY and JSON column types, and also composite column as stated in documentation - Composite Column Types. All of which are failed as follows.

• composite is not working

  d1 = Column(Boolean)
  d2 = Column(Boolean)
  type = composite(Discriminator, d1, d2)
  __mapper_args__ = {
      "polymorphic_identity": Discriminator(False, False),
      "polymorphic_on": type,
  }
  ...
  ---
  Error:
      sqlalchemy.exc.ArgumentError: Column expression or string key expected for argument 'polymorphic_on'; got <CompositeProperty at 0xffff9ec70be0; no key>.

• dict is not working:

  type = Column(JSON)
  __mapper_args__ = {
      "polymorphic_identity": {"d1": False, "d2": False},
      "polymorphic_on": type,
  }
  ...
  ---
  Error:
    File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/mapper.py", line 689, in __init__
      self._configure_inheritance()
    File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/mapper.py", line 1151, in _configure_inheritance
      self.polymorphic_map[self.polymorphic_identity] = self
  TypeError: unhashable type: 'dict'

• list is not working too, with unhashable type error similar as dict

Then I dig deep into internet, found that #8089 are quite similar to my case. And I implemented an example below

from typing import Tuple

from sqlalchemy import (
    create_engine,
    Column,
    ForeignKey,
    Integer,
    select,
    inspect,
    cast,
    event,
    String,
    Boolean,
)
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()


class ident_(str):

    """describe a composed identity.

    Using a string for easy conversion to a string SQL composition.
    """

    _tup: Tuple[bool, bool]

    def __new__(cls, d1, d2):
        self = super().__new__(cls, f"{d1}{d2}")
        self._tup = d1, d2
        return self

    def _as_tuple(self):
        return self._tup


class Person(Base):
    __tablename__ = "person"
    id: Column = Column(Integer, primary_key=True)
    d1 = Column(Boolean, nullable=False)  # this can be your FK to the other table etc.
    d2 = Column(Boolean, nullable=True)  # this is a "supplementary" discrim column

    __mapper_args__ = {
        "polymorphic_identity": ident_(False, False),
        "polymorphic_on": cast(d1, String) + cast(d2, String),
    }


@event.listens_for(Person, "init", propagate=True)
def _setup_poly(target, args, kw):
    """receive new AbstractQuestion objects when they are constructed and
    set polymorphic identity"""

    # this is the ident_() object
    ident = inspect(target).mapper.polymorphic_identity

    d1, d2 = ident._as_tuple()
    kw["d1"] = d1
    kw["d2"] = d2


class Employee(Person):
    __tablename__ = "employee"
    id: Column = Column(Integer, ForeignKey(Person.id), primary_key=True)

    __mapper_args__ = {"polymorphic_identity": ident_(True, False)}


class Student(Person):
    __tablename__ = "student"
    id: Column = Column(Integer, ForeignKey(Person.id), primary_key=True)

    __mapper_args__ = {"polymorphic_identity": ident_(False, True)}


engine = create_engine("sqlite://", echo=True, future=True)

Base.metadata.create_all(engine)

Session = sessionmaker(engine)

with Session() as session:
    # adding a person to each table
    id_ = 1
    session.add(Employee(id=id_))
    session.commit()
    # search for inserted person
    person_id = (
        session.execute(select(Person.id).where(Person.id == id_))
        .scalars()
        .one_or_none()
    )
    employee_id = (
        session.execute(select(Employee.id).where(Employee.id == id_))
        .scalars()
        .one_or_none()
    )
    session.commit()
    # check if their ids equal
    assert person_id == employee_id == id_

    # try add the same person to Student subclass
    session.add(Student(id=id_))
    session.commit()
    # Oops, IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: person.id

But then I found no clues to continue implementing overlapping joined table inheritance due to IntegrityError. Could anybody give me some hints? Is there any possibility to implement this with the polymorphic functionality stated in documentation - Mapping Class Inheritance Hierarchies?

Or is the following code the only way to implement overlapping joined table inheritance? i.e. without the polymorphic functionality and do joining and inserting manually?

from sqlalchemy import create_engine, Column, ForeignKey, Integer, select
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()


class Person(Base):
    __tablename__ = "person"
    id: Column = Column(Integer, primary_key=True)


class Employee(Base):
    __tablename__ = "employee"
    id: Column = Column(Integer, ForeignKey(Person.id), primary_key=True)


class Student(Base):
    __tablename__ = "student"
    id: Column = Column(Integer, ForeignKey(Person.id), primary_key=True)


engine = create_engine("sqlite://", echo=True, future=True)

Base.metadata.create_all(engine)

Session = sessionmaker(engine)

with Session() as session:
    # adding a person to each table
    id_ = 1
    session.add(Person(id=id_))
    session.add(Employee(id=id_))
    session.add(Student(id=id_))
    session.commit()
    # search for inserted person
    person_id = (
        session.execute(select(Person.id).where(Person.id == id_))
        .scalars()
        .one_or_none()
    )
    employee_id = (
        session.execute(select(Employee.id).where(Employee.id == id_))
        .scalars()
        .one_or_none()
    )
    student_id = (
        session.execute(select(Student.id).where(Student.id == id_))
        .scalars()
        .one_or_none()
    )
    session.commit()
    # check if their ids equal
    assert person_id == employee_id == student_id == id_
    # No error, perfectly fine

[CaselIT]

Hi.

I may be wrong here, but I don't think you can do this without a 3rd model StudentEmployee here.
How you are trying to model things does not seem correct to me, since by definition a person is either a student or an employee.

Personally in this case I think I would model the data differently to have a Person table and separate Student and Employment tables that hold information of about a person

[benlau6]

And a person could be a student and an employee at the same time because I was a full time employee and also a part time student years ago, though it is probably an edge case. I am sorry that I provided a hypothetical case that is too minimal and not comprehensive enough. Please let me give out my real business scenario.

Note that I was not doubting the validity of the example, just that for modelling purposes with just two classes Student and Employee you cannot have a person that is a stent and an employee, as mike pointed out.

To put it practically, what would session.get(Person, 1) return in the hypothetical case that 1 is the id of a person that's both a student and an employee?

Sorry for misunderstanding and to clarify it so late that I will never have such query. I just need him/her to be either a student or an employee while querying, but I wanna store it into database without violating unique key constraint, and make good use of inheritance to gather all the related information I need while querying him/her as a Employee, or sometimes querying him/her as a Student

[zzzeek]

OK so what kind of query do you envision, and what kind of object would you get back? Be specific.

A query you might do is:

object = session.get(Employee, 1)

right? now if "object" in fact is a "Student" and an "Employee" at the same time, what would type(object) return? What we are saying is that it cannot be Employee, since Employee is not Student. It can only be a class that expresses both, SomeClass(Employee, Student).

check out https://en.wikipedia.org/wiki/Multiple_inheritance for background.

[benlau6]

Thank you for your patience.

Let's say I have a request service. There are clients who want to request service, and I have two types of service providers, which are Employee and Student. At first, a client need to declare what type of service he/she need. So there are only two cases

1. A client needs a service that can only be handled by Employee. Let's check what candidates I have

   candidates = session.execute(Employee).scalars().all()
   

2. A client needs a service that can only be handled by Student. Let's check what candidates I have

   candidates = session.execute(Student).scalars().all()
   

From querying perspective, and object perspective, that's all. I want either Employee or Student, but not both, in any request.

But I have a person 'Adam' who is in fact employee and student. It's "Person" related attributes (name) need to be stored in Person table, its "Employee" related attribute (job_title) need to be stored in Employee table. That's great. I can do it by

adam = Employee(name="adam", job_title="intern")
session.add(adam)

But then I have no clue how to also store its "Student" related attribute (major) in Student table. If I do it as follows, it will raise IntegrityError since there are two instances from Person table with the same id.

# add Adam into `Person` and `Employee` tables
adam = Employee(name="adam", job_title="intern") # done in previous step
session.add(adam) # done in previous step
session.flush()
# find Adam's inserted id, add Adam into `Student` table
adam_id = adam.id
adam_student = Student(id=adam_id, major='Computer Science')
session.add(adam_student)
session.commit()

[zzzeek]

But I have a person 'Adam' who is in fact employee and student. It's "Person" related attributes (name) need to be stored in Person table, its "Employee" related attribute (job_title) need to be stored in Employee table. That's great. I can do it by

adam = Employee(name="adam", job_title="intern")
session.add(adam)

But then I have no clue how to also store its "Student" related attribute (major) in Student table. If I do it as follows, it will raise IntegrityError since there are two instances from Person table with the same id.

# add Adam into `Person` and `Employee` tables
adam = Employee(name="adam", job_title="intern") # done in previous step
session.add(adam) # done in previous step
session.flush()
# find Adam's inserted id, add Adam into `Student` table
adam_id = adam.id
adam_student = Student(id=adam_id, major='Computer Science')
session.add(adam_student)
session.commit()

Yes sorry, that's not how the feature works. A particular primary key identity only loads a single type of object, so if Person has a primary key "1", that can join only to Employee or Student, not both.

So foregoing the use of the inheritance feature entirely, you here could map Employee to a join of "employee" and "person" table, and then map Student to a join of "student" and "person" table, the Person superclass would not be mapped at all (it would just be a common mixin), and you would use two independent mappings. Your Session could then have two different objects for "adam" at once, that is, you would load Employee(name="adam") and Student(name="adam") separately. However, mapping to a join directly (described at https://docs.sqlalchemy.org/en/14/orm/nonstandard_mappings.html#mapping-a-class-against-multiple-tables ) doesn't have all the "polymorphic" features of inheritance - you just would deal with Employee or Student directly for any particular query. you also would not need any "discriminator" column since you know ahead of time that you are querying for Employee or Student and you would query for the specific JOIN that you already know you want.

[benlau6]

It is clear to me now.
Sorry for making the discussion going round in circles. I should state the circumstance more clear in advance.
Thank you, zzzeek and CaselIT, for your patient and prompt assistance.

Below is a minimal example for future reference

from sqlalchemy import (
    String,
    create_engine,
    Column,
    ForeignKey,
    Integer,
    select,
    join,
    update,
)
from sqlalchemy.orm import declarative_base, sessionmaker, column_property

Base = declarative_base()


class Person(Base):
    __tablename__ = "person"
    id: Column = Column(Integer, primary_key=True)
    name = Column(String)


class Employee(Base):
    __tablename__ = "employee"
    id: Column = Column(Integer, ForeignKey(Person.id), primary_key=True)
    title = Column(String)


class Student(Base):
    __tablename__ = "student"
    id: Column = Column(Integer, ForeignKey(Person.id), primary_key=True)
    major = Column(String)


class EmployeePerson(Base):
    __table__ = join(Person, Employee)
    id = column_property(Person.id, Employee.id)


class StudentPerson(Base):
    __table__ = join(Person, Student)
    id = column_property(Person.id, Student.id)


engine = create_engine("sqlite://", echo=True, future=True)

Base.metadata.create_all(engine)

Session = sessionmaker(engine)

with Session() as session:
    # add Adam to each table
    id_ = 1
    name = "Adam"
    title = "internship"
    major = "computer science"
    session.add(EmployeePerson(id=id_, name=name, title=title))
    session.add(Student(id=id_, major=major))
    session.commit()

    # query the inserted instances
    person = (
        session.execute(
            select(Person)
            .where(Person.id == id_)
            .where(Person.name == name)
        )
        .scalars()
        .one_or_none()
    )
    employee = (
        session.execute(
            select(EmployeePerson)
            .where(EmployeePerson.id == id_)
            .where(EmployeePerson.name == name)
        )
        .scalars()
        .one_or_none()
    )
    student = (
        session.execute(
            select(StudentPerson)
            .where(StudentPerson.id == id_)
            .where(StudentPerson.name == name)
        )
        .scalars()
        .one_or_none()
    )
    session.commit()
    # check if their attributes equal as expected
    assert person.id == employee.id == student.id == id_
    assert person.name == employee.name == student.name == name
    assert employee.title == title
    assert student.major == major

    # check if updating is also fine
    session.execute(update(Person).where(Person.name == name).values(name="Robert"))
    session.commit()
    assert person.name == employee.name == student.name == "Robert"