python - order_by - sqlalchemy relationship example

SQLAlchemy: no aplique la restricción de clave externa en una relación

Question

Tengo un modelo / tabla de Test y un modelo / tabla TestAuditLog , usando SQLAlchemy y SQL Server 2008. La relación entre los dos es Test.id == TestAuditLog.entityId , con una prueba que tiene muchos registros de auditoría. TestAuditLog está destinado a mantener un historial de cambios en las filas en la tabla de Test . También quiero rastrear cuándo se elimina una Test , pero estoy teniendo problemas con esto. En SQL Server Management Studio, establecí la FK_TEST_AUDIT_LOG_TEST " Imponer restricción de clave externa " de la relación FK_TEST_AUDIT_LOG_TEST a "No", pensando que permitiría que TestAuditLog una fila TestAuditLog con un entityId que ya no se conecta a ningún Test.id porque la Test se ha eliminado . Sin embargo, cuando intento crear un TestAuditLog con SQLAlchemy y luego eliminar la Test , TestAuditLog un error:

(IntegrityError) (''23000'', "[23000] [Microsoft] [Controlador ODBC SQL Server] [SQL Server] No se puede insertar el valor NULL en la columna ''AL_TEST_ID'', la columna ''TEST_AUDIT_LOG''; la columna no permite valores nulos. ACTUALIZACIÓN falla. (515) (SQLExecDirectW); [01000] [Microsoft] [Controlador ODBC de SQL Server] [SQL Server] La instrucción ha finalizado. (3621) ") u''UPDATE [TEST_AUDIT_LOG] SET [AL_TEST_ID] =? DONDE [TEST_AUDIT_LOG]. [AL_ID] =? '' (Ninguno, 8)

Creo que debido a la relación de clave externa entre Test y TestAuditLog , después de eliminar la fila Test , SQLAlchemy intenta actualizar todos los registros de auditoría de la prueba para tener un entityId NULL . No quiero que haga esto; Quiero que SQLAlchemy deje los registros de auditoría solo. ¿Cómo puedo decirle a SQLAlchemy que permita que existan registros de auditoría cuya entityId no se conecte con ningún Test.id ?

Intenté simplemente eliminar ForeignKey de mis tablas, pero me gustaría seguir pudiendo decir myTest.audits y obtener todos los registros de auditoría de una prueba, y SQLAlchemy se quejó de no saber cómo unirse a Test y TestAuditLog . Cuando especifiqué un primaryjoin en la relationship , se quejó de no tener una ForeignKey o ForeignKeyConstraint con las columnas.

Aquí están mis modelos:

class TestAuditLog(Base, Common):
    __tablename__ = u''TEST_AUDIT_LOG''
    entityId = Column(u''AL_TEST_ID'', INTEGER(), ForeignKey(u''TEST.TS_TEST_ID''),
        nullable=False)
    ...

class Test(Base, Common):
    __tablename__ = u''TEST''
    id = Column(u''TS_TEST_ID'', INTEGER(), primary_key=True, nullable=False)
    audits = relationship(TestAuditLog, backref="test")
    ...

Y así es como estoy tratando de eliminar una prueba mientras entityId sus registros de auditoría, su entityId intacto:

    test = Session.query(Test).first()
    Session.begin()
    try:
        Session.add(TestAuditLog(entityId=test.id))
        Session.flush()
        Session.delete(test)
        Session.commit()
    except:
        Session.rollback()
        raise

Answer 1

Puedes resolver esto por:

• POINT-1: no tiene una ForeignKey ni en el nivel RDBMS ni en el nivel SA
• POINT-2: especifique explícitamente condiciones de unión para la relación
• PUNTO-3: marque las cascadas de relación para confiar en la bandera de los indicadores pasivos

El fragmento de código que funciona a continuación debería darte una idea (los puntos se destacan en el code ):

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import scoped_session, sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = create_engine(''sqlite:///:memory:'', echo=False)

Session = sessionmaker(bind=engine)

class TestAuditLog(Base):
    __tablename__ = ''TEST_AUDIT_LOG''
    id = Column(Integer, primary_key=True)
    comment = Column(String)

    entityId = Column(''TEST_AUDIT_LOG'', Integer, nullable=False,
                     # POINT-1
                     #ForeignKey(''TEST.TS_TEST_ID'', ondelete="CASCADE"),
                     )

    def __init__(self, comment):
        self.comment = comment

    def __repr__(self):
        return "<TestAuditLog(id=%s entityId=%s, comment=%s)>" % (self.id, self.entityId, self.comment)

class Test(Base):
    __tablename__ = ''TEST''
    id = Column(''TS_TEST_ID'', Integer, primary_key=True)
    name = Column(String)

    audits = relationship(TestAuditLog, backref=''test'',
                # POINT-2
                primaryjoin="Test.id==TestAuditLog.entityId",
                foreign_keys=[TestAuditLog.__table__.c.TEST_AUDIT_LOG],
                # POINT-3
                passive_deletes=''all'',
            )

    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return "<Test(id=%s, name=%s)>" % (self.id, self.name)


Base.metadata.create_all(engine)

###################
## tests
session = Session()

# create test data
tests = [Test("test-" + str(i)) for i in range(3)]
_cnt = 0
for _t in tests:
    for __ in range(2):
        _t.audits.append(TestAuditLog("comment-" + str(_cnt)))
        _cnt += 1
session.add_all(tests)
session.commit()
session.expunge_all()
print ''-''*80

# check test data, delete one Test
t1 = session.query(Test).get(1)
print "t: ", t1
print "t.a: ", t1.audits
session.delete(t1)
session.commit()
session.expunge_all()
print ''-''*80

# check that audits are still in the DB for deleted Test
t1 = session.query(Test).get(1)
assert t1 is None
_q = session.query(TestAuditLog).filter(TestAuditLog.entityId == 1)
_r = _q.all()
assert len(_r) == 2
for _a in _r:
    print _a

Otra opción sería duplicar la columna utilizada en el FK, y hacer que la columna FK sea nulable con la opción ON CASCADE SET NULL . De esta forma, aún puede verificar la pista de auditoría de los objetos eliminados utilizando esta columna.