Cheers
Simon
On Tuesday, 6 December 2016 14:36:15 UTC, dcs3spp wrote:
Hi,
I am running SQLAlchemy 1.1.4 connected to postgreSQL 9.6 on OSX
Sierra 10.12.1. I am experiencing difficulty getting SQLAlchemy to
raise an integrity error when a persistent object is appended to a
relationship for which it is already associated with.
Is the accepted method to manually raise an exception by testing for
existing membership in the collection_class of the relationship
before inserting the persistent object into the relationship? I have
tested with default list and also setting the collection_class to
set. Would be grateful if anyone could help with this and point out
where I am going wrong :)
I have included details of code and DB structure below. Currently,
duplicate associations are silently ignored, i.e. no INSERT
statements generated.
Cheers
Simon
DB Structure
Assignment —< AssignmentOutcome >— LearningOutcome
TABLE KEYS
Assignment identified by AssignmentID
LearningOutcome identified by CourseID, UnitID, OutcomeID
AssignmentOutcome identified by AssignmentID, CourseID, UnitID,
OutcomeID
ASSOCIATION TABLE STATE BEFORE TEST SCRIPT RUN
proposed_coursemanagement=> SELECT * FROM assignment outcome;
AssignmentID | CourseID | UnitID | OutcomeID
--------------+----------+--------+-----------
1 | 1014760 | 2 | 1
1 | 1014760 | 2 | 2
2 | 1014760 | 2 | 1
2 | 1014760 | 2 | 3
3 | 1014760 | 2 | 2
4 | 1014760 | 2 | 3
(6 rows)
TEST SCRIPT PURPOSE: Try appending outcome 1 again to assignment 1.
Should SQLAlchemy raise and exception since this is already mapped
in association table? Current behaviour is that duplicate is
silently ignored. When I append outcome 4 for assignment 1 it is
mapped to the association table.
TEST SCRIPT CODE:
def usage(argo):
cmd = os.path.basename(argv[0])
print('usage: %s <config_uri> [var=value]\n'
'(example: "%s development.ini")' % (cmd, cmd))
def main(argv=sys.argv):
if len(argv) < 2:
usage(argv)
print ("Initialising SQLAlchemy engine instance and tables.......")
config_uri = argv[1]
options = parse_vars(argv[2:])
setup_logging(config_uri)
settings = get_appsettings(config_uri, options=options)
engine = get_engine(settings)
Base.metadata.create_all(engine)
session_factory = get_session_factory(engine)
print ("Completed initialisation.....")
print ("Starting Test Script...")
with transaction.manager:
dbsession = get_tm_session(session_factory, transaction.manager)
# load into session assignment 1 and learning outcomes 1 and 4
# outcome 1 already exists in the association table for
assignment 1
# outcome 4 does not exist yet in the association table for
assignment 1
assignment = dbsession.query(AssignmentModel).get(1)
outcome_1 =
dbsession.query(LearningOutcomeModel).get((1014760, 2, 1))
outcome_4 =
dbsession.query(LearningOutcomeModel).get((1014760, 2, 4))
# add outcome 1 and outcome 4 to assignmentoutcomes relationship
assignment.assignmentoutcomes.append(outcome_1)
assignment.assignmentoutcomes.append(outcome_4)
# inspect and display state for assignment and outcome objects
assignmentIns = inspect (assignment)
outcome1Ins = inspect (outcome_1)
outcome4Ins = inspect (outcome_4)
print("assignment object=Transient: {0}, Pending: {1},
Persistent: {2}, Detatched: {3}".format(
assignmentIns.transient, assignmentIns.pending,
assignmentIns.persistent, assignmentIns.detached))
print("outcome_1 object=Transient: {0}, Pending: {1},
Persistent: {2}, Detatched: {3}".format(
outcome1Ins.transient, outcome1Ins.pending,
outcome1Ins.persistent, outcome1Ins.detached))
print("outcome_4 object=Transient: {0}, Pending: {1},
Persistent: {2}, Detatched: {3}".format(
outcome4Ins.transient, outcome4Ins.pending,
outcome4Ins.persistent, outcome4Ins.detached))
# inspect and display session states .dirty, .new and .deleted
print ("dbsession.dirty={0}, .new={1}, .deleted={2}".format(
dbsession.dirty, dbsession.new, dbsession.deleted))
print ("Test Script complete...")
TEST SCRIPT OUTPUT
assignment object=Transient: False, Pending: False, Persistent:
True, Detatched: False
outcome_1 object=Transient: False, Pending: False, Persistent: True,
Detatched: False
outcome_4 object=Transient: False, Pending: False, Persistent: True,
Detatched: False
dbsession.dirty=IdentitySet([<LearningOutcome('1014760','2', '4',
'Unit 2 : Learning Outcome 4')>, <LearningOutcome('1014760','2',
'1', 'Understand the components of computer systems')>,
<Assignment('1','1014760', '2', '2015-12-10 00:00:00+00:00',
'Assignment 1')>]), .new=IdentitySet([]), .deleted=IdentitySet([])
2016-12-06 12:33:10,691 INFO
[sqlalchemy.engine.base.Engine:1097][MainThread] INSERT INTO
proposed_coursemanagement.assignmentoutcome ("CourseID", "UnitID",
"OutcomeID", "AssignmentID") VALUES (%(CourseID)s, %(UnitID)s,
%(OutcomeID)s, %(AssignmentID)s)
2016-12-06 12:33:10,691 INFO
[sqlalchemy.engine.base.Engine:1100][MainThread] {'OutcomeID': 4,
'CourseID': 1014760, 'UnitID': 2, 'AssignmentID': 1}
2016-12-06 12:33:10,692 INFO
[sqlalchemy.engine.base.Engine:686][MainThread] COMMIT
Test Script complete...
ASSOCIATION TABLE STATE AFTER TEST SCRIPT RUN
proposed_coursemanagement=> SELECT * FROM assignmentoutcome;
AssignmentID | CourseID | UnitID | OutcomeID
--------------+----------+--------+-----------
1 | 1014760 | 2 | 1 * No duplicate raised
- does SQLAlchemy silently ignore?
1 | 1014760 | 2 | 2
2 | 1014760 | 2 | 1
2 | 1014760 | 2 | 3
3 | 1014760 | 2 | 2
4 | 1014760 | 2 | 3
1 | 1014760 | 2 | 4
(7 rows)
MODEL CODE:
assignmentoutcomeallocation = Table('assignmentoutcome', Base.metadata,
Column('CourseID', BigInteger, primary_key=True),
Column('UnitID', Integer, primary_key=True),
Column('OutcomeID', Integer, primary_key=True),
Column('AssignmentID', BigInteger,
ForeignKey('assignment.AssignmentID'), primary_key=True),
UniqueConstraint('CourseID', 'UnitID', 'OutcomeID', 'AssignmentID'),
ForeignKeyConstraint(
['CourseID', 'UnitID', 'OutcomeID'],
['learningoutcome.CourseID', 'learningoutcome.UnitID',
'learningoutcome.OutcomeID']),
)
class AssignmentModel(Base):
__tablename__ = 'assignment'
__table_args__ = (
ForeignKeyConstraint(['CourseID','UnitID'],
['unit.CourseID', 'unit.UnitID']),
)
AssignmentID = Column(BigInteger, primary_key=True,
server_default=text("nextval('proposed_coursemanagement.\"assignment_AssignmentID_seq\"::regclass)"))
CourseID = Column(BigInteger, nullable=False)
UnitID = Column(Integer, nullable=False)
AssignmentSetDate = Column(DateTime(timezone=True), nullable=False)
AssignmentSubmissionDate = Column(DateTime(timezone=True),
nullable=False)
AssignmentResubmissionDate = Column(DateTime(timezone=True),
nullable=False)
AssignmentTitle = Column(String(200), nullable=False)
indx = Index('assignment_indx', 'CourseID', 'UnitID',
text('lower(\"AssignmentTitle\")'), 'AssignmentSetDate')
unit = relationship('UnitModel')
assignmentcriteria = relationship('UnitCriteriaModel',
secondary=assignmentcriteriaallocation,
back_populates='criteria_assignments')
assignmentoutcomes = relationship('LearningOutcomeModel',
secondary=assignmentoutcomeallocation,
back_populates='outcome_assignments')
def __repr__(self):
return "<Assignment('%s','%s', '%s', '%s', '%s')>" %
(self.AssignmentID, self.CourseID, self.UnitID,
self.AssignmentSetDate, self.AssignmentTitle)
def __str__(self):
return "<Assignment('%s','%s', '%s', '%s', '%s')>" %
(self.AssignmentID, self.CourseID, self.UnitID,
self.AssignmentSetDate, self.AssignmentTitle)
class LearningOutcomeModel(Base):
__tablename__ = 'learningoutcome'
__table_args__ = (
ForeignKeyConstraint(['CourseID','UnitID'],
['unit.CourseID', 'unit.UnitID']),
)
CourseID = Column(BigInteger, primary_key=True, nullable=False)
UnitID = Column(Integer, primary_key=True, nullable=False)
OutcomeID = Column(Integer, primary_key=True, nullable=False)
LearningOutcome = Column(String(100), nullable=False)
unit = relationship('UnitModel')
outcome_assignments = relationship (
'AssignmentModel',
secondary=assignmentoutcomeallocation,
back_populates='assignmentoutcomes')
def __repr__(self):
return "<LearningOutcome('%s','%s', '%s', '%s')>" %
(self.CourseID, self.UnitID, self.OutcomeID, self.LearningOutcome)
def __str__(self):
return "<LearningOutcome('%s','%s', '%s', '%s')>" %
(self.CourseID, self.UnitID, self.OutcomeID, self.LearningOutcome)
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and
Verifiable Example. See http://stackoverflow.com/help/mcve for a full
description.
---
You received this message because you are subscribed to the Google
Groups "sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send
an email to [email protected]
<mailto:[email protected]>.
To post to this group, send email to [email protected]
<mailto:[email protected]>.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
