How can I define a constraint on an inherited column in SQLAlchemy?

Question

I have a class inheritance scheme as layed out in http://docs.sqlalchemy.org/en/latest/orm/inheritance.html#joined-table-inheritance and I'd like to define a constraint that uses columns from both the parent and child classes.

from sqlalchemy import (
    create_engine, Column, Integer, String, ForeignKey, CheckConstraint
)
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()


class Parent(Base):
    __tablename__ = 'parent'

    id = Column(Integer, primary_key=True)
    type = Column(String)
    name = Column(String)

    __mapper_args__ = {'polymorphic_on': type}


class Child(Parent):
    __tablename__ = 'child'

    id = Column(Integer, ForeignKey('parent.id'), primary_key=True)
    child_name = Column(String)

    __mapper_args__ = {'polymorphic_identity': 'child'}
    __table_args__ = (CheckConstraint('name != child_name'),)


engine = create_engine(...)
Base.metadata.create_all(engine)

This doesn't work because name isn't a column in child; I get the error

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) column "name" does not exist
 [SQL: '\nCREATE TABLE child (\n\tid INTEGER NOT NULL, \n\tPRIMARY KEY (id), \n\tCHECK (name="something"), \n\tFOREIGN KEY(id) REFERENCES parent (id)\n)\n\n']

So how can I define such a constraint?

Answer 1

After some tinkering I came up with a solution: create a "copy" parent_name column in Child that references name in Parent. It wastes some storage space, but that's probably unavoidable in order to have a real CHECK CONSTRAINT.

Here is the code:

from sqlalchemy import (
    create_engine, Column, Integer, String,
    CheckConstraint, UniqueConstraint, ForeignKeyConstraint
)
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.exc import IntegrityError


Base = declarative_base(metadata=metadata)


class Parent(Base):
    __tablename__ = 'parent'

    id = Column(Integer, primary_key=True)
    type = Column(String, nullable=False)
    name = Column(String, nullable=False)

    __mapper_args__ = {'polymorphic_on': type}
    __table_args__ = (UniqueConstraint('id', 'name'),)


class Child(Parent):
    __tablename__ = 'child'

    id = Column(Integer, primary_key=True)
    parent_name = Column(String, nullable=False)
    child_name = Column(String, nullable=False)

    __mapper_args__ = {'polymorphic_identity': 'child'}
    __table_args__ = (
        ForeignKeyConstraint(
            ['id', 'parent_name'], ['parent.id', 'parent.name'],
            onupdate='CASCADE', ondelete='CASCADE'
        ),
        CheckConstraint('parent_name != child_name'),
    )


engine = create_engine(...)
Base.metadata.create_all(engine)
session = sessionmaker(bind=engine, autocommit=True)()

print('Works without error:')
print('--------------------')

with session.begin():
    session.add(Child(name='a', child_name='b'))

print(session.query(Child).one().__dict__)

with session.begin():
    child = session.query(Child).one()
    child.name = 'c'

print(session.query(Child).one().__dict__)

print('\nFails due to IntegerityError:')
print('-------------------------------')
try:
    with session.begin():
        session.add(Child(name='a', child_name='a'))
except IntegrityError as e:
    print(e.orig)
    print(e.statement)

The output of this script is

Works without error:
--------------------
{'type': 'child', '_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7f5e9b9b7898>, 'id': 1, 'child_name': 'b', 'parent_name': 'a', 'name': 'a'}
{'type': 'child', '_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7f8fc80f2b38>, 'id': 1, 'child_name': 'b', 'parent_name': 'c', 'name': 'c'}

Fails due to IntegerityError:
-------------------------------
new row for relation "child" violates check constraint "child_check"
DETAIL:  Failing row contains (2, a, a).

INSERT INTO child (id, parent_name, child_name) VALUES (%(id)s, %(parent_name)s, %(child_name)s)

Answer 2

Simple answer: you cannot do this using CHECK Constraint.

You cannot do it in the plain RDBMS, therefore you cannot do in using SQLAlchemy.
However, if all the data modifications are passing through your application (and not with direct DB access), you can add validation routines to your classes:

class Child(Parent):
    # ...

    @validates('child_name')
    def validate_child_name(self, key, child_name):
        assert child_name != name
        return child_name

Read more Simple Validators.