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.