Last active
June 23, 2024 20:59
-
-
Save jdittrich/3001e520d3872b12e2cb8e7d4a2472da to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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