Hi Mike,
Thanks for getting back to me and suggesting viable workarounds :) Will
probably go with the event listener approach.
Is it worthwhile submitting this as an issue on bitbucket?
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 sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
