Skip to content

Instantly share code, notes, and snippets.

@jdittrich
Last active June 23, 2024 20:59
Show Gist options
  • Save jdittrich/3001e520d3872b12e2cb8e7d4a2472da to your computer and use it in GitHub Desktop.
Save jdittrich/3001e520d3872b12e2cb8e7d4a2472da to your computer and use it in GitHub Desktop.
from sqlalchemy import create_engine, Column, Table, ForeignKey, Integer, String
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import Mapped
engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
session = Session()
Base = declarative_base()
def create_tables():
Base.metadata.create_all(bind=engine)
# based on https://stackoverflow.com/a/21670302/263398
class Person(Base):
__tablename__ = 'person'
id: Mapped[int]= mapped_column(Integer, primary_key=True)
name:Mapped[str] = mapped_column(String(50))
parties:Mapped[list["Party"]] = relationship(secondary='guest_association', viewonly=True ) # viewonly prevents an SAWarning on overlap (https://docs.sqlalchemy.org/en/20/errors.html#error-qzyx)
class Party(Base):
__tablename__ = 'party'
id = mapped_column(Integer, primary_key=True)
# note: a backref adds the attribute on the targeted other table.
# but in modern Python, explicit use of relationship() with relationship.back_populates should be preferred,
# as it is more robust in terms of mapper configuration as well as more conceptually straightforward.
# parameter "secondary" does define the association table
guests:Mapped[list["Person"]] = relationship(
'Person',
secondary='guest_association',
back_populates='parties',
viewonly=True
)
# viewonly prevents an SAWarning on overlap (https://docs.sqlalchemy.org/en/20/errors.html#error-qzyx)
# deleted lazy=dynamic from original example
association_recs:Mapped[list["GuestAssociation"]] = relationship("GuestAssociation", back_populates="party", cascade="all, delete-orphan")
# we need the cascade, since otherwise the entries will not be deleted ("Dependency rule tried to blank-out primary key column")
association_ids:Mapped[list[int]] = association_proxy(
"association_recs", "user_id",
creator=lambda uid: GuestAssociation(user_id=uid))
guest_association = Table(
'guest_association', Base.metadata,
Column('user_id', Integer(), ForeignKey('person.id'), primary_key=True),
Column('party_id', Integer(), ForeignKey('party.id'), primary_key=True)
)
class GuestAssociation(Base):
__table__ = guest_association
party:Mapped["Party"] = relationship('Party',back_populates="association_recs")
create_tables()
# try this:
bob = Person(id=1, name='bob')
jane = Person(id=2, name='jane')
linda = Person(id=3, name='linda')
john = Person(id=4, name='john')
sp1 = Party(id=1) #id was 1
session.add_all([sp1, bob, jane, linda, john])
session.commit()
sp1.association_ids.extend([3, 4])
session.commit()
#did it work?
# linda (id=3) and john (id=4) should be at the party with the id 1
print(linda.parties[0].id) # 1
print(john.parties[0].id) # 1
# bob and jane are at no party
print("count of parties:", len(bob.parties)) # count of parties: 0
print("count of parties:", len(jane.parties))# count of parties: 0
# kick out john
indexOfJohn = sp1.association_ids.index(4)
sp1.association_ids.pop(indexOfJohn)
session.commit()
print("count of parties:", len(john.parties))# count of parties: 0
# hope we did not cascade anything away; party still there
print(linda.parties[0].id) # 1
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment