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

I have a class inheritance schema as linked at http://docs.sqlalchemy.org/en/latest/orm/inheritance.html#joined-table-inheritance and I would like to define a constraint that uses columns as 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)

      

It doesn't work because it is name

not a column in child

; I get an 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?

+3


source to share


2 answers


After some tinkering, I came up with a solution: create a "copy" column parent_name

in Child

that refers to name

in Parent

. It takes up some storage space, but it's probably inevitable to have a real one 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)

      

+1


source


Simple answer: you cannot do this with a CHECK constraint.

You cannot do this in a simple RDBMS, so you cannot use SQLAlchemy.
However, if all data changes go through your application (and not with direct database access), you can add validation routines for 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 .

+1


source







All Articles