Created
October 26, 2022 21:16
-
-
Save brandonrobertz/96536b88698887df0e4219860d4cceac to your computer and use it in GitHub Desktop.
SQLAlchemy Quickstart Annotated
This file contains 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 | |
from sqlalchemy import select | |
from sqlalchemy import Column | |
from sqlalchemy import ForeignKey | |
from sqlalchemy import Integer | |
from sqlalchemy import String | |
from sqlalchemy.orm import declarative_base | |
from sqlalchemy.orm import relationship | |
from sqlalchemy.orm import Session | |
Base = declarative_base() | |
class User(Base): | |
""" | |
User model. This will become a table in our DB called user_account | |
""" | |
__tablename__ = "user_account" | |
# these are columns in the user_account table | |
id = Column(Integer, primary_key=True) | |
name = Column(String(30)) | |
fullname = Column(String) | |
# each user can have one or more addresses associated | |
addresses = relationship( | |
# user.addresses will be a list of instances of Address model (below) | |
"Address", | |
# this means the address model will have a user, leading back | |
# to the user who is associated with the address | |
back_populates="user", | |
# deleting this user will cause all associated addressed to be deleted | |
# as long as they're not used by other users | |
cascade="all, delete-orphan" | |
) | |
def __repr__(self): | |
""" | |
When we print an instance of the User class, this will be printed. | |
""" | |
return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})" | |
class Address(Base): | |
__tablename__ = "address" | |
id = Column(Integer, primary_key=True) | |
email_address = Column(String, nullable=False) | |
user_id = Column(Integer, ForeignKey("user_account.id"), nullable=False) | |
# each address is associated with one user | |
user = relationship("User", back_populates="addresses") | |
def __repr__(self): | |
return f"Address(id={self.id!r}, email_address={self.email_address!r})" | |
if __name__ == "__main__": | |
# SQLite memory-only database for convenience | |
engine = create_engine( | |
# if you want the DB to be written to disk change | |
# this to: sqlite:///quickstart.db | |
"sqlite:///:memory:", | |
# print out SQL statements | |
echo=False, | |
# use only sqlalchemy 2.0 APIs | |
future=True | |
) | |
# create all DBs | |
Base.metadata.create_all(engine) | |
# insert some rows, using a transaction | |
with Session(engine) as session: | |
spongebob = User( | |
name="spongebob", | |
fullname="Spongebob Squarepants", | |
addresses=[ | |
# Create and add relationship from spongebob to this email | |
Address(email_address="spongebob@sqlalchemy.org"), | |
Address(email_address="shared@spongebob.org"), | |
], | |
) | |
print("Adding user", spongebob) | |
sandy = User( | |
name="sandy", | |
fullname="Sandy Cheeks", | |
addresses=[ | |
# create and add these emails | |
Address(email_address="sandy@sqlalchemy.org"), | |
Address(email_address="sandy@squirrelpower.org"), | |
Address(email_address="shared@spongebob.org"), | |
], | |
) | |
print("Adding user", sandy) | |
patrick = User( | |
name="patrick", fullname="Patrick Star", | |
addresses=[ | |
Address(email_address="shared@spongebob.org"), | |
] | |
) | |
print("Adding user", patrick) | |
# prepare these changes writing ... | |
session.add_all([spongebob, sandy, patrick]) | |
# ... this actually writes the changes to the DB | |
session.commit() | |
# query for sandy, by name | |
stmt = select(User).where(User.name == "sandy") | |
# get sandy user (the first result of the sandy query above) | |
sandy = session.scalars(stmt).one() | |
print("Sandy", sandy) | |
# get patrick's email addresses. this will go into the | |
# addresses table and pull out patrick's email addys | |
# using the relationship defined in the models above | |
print("Sandy has", len(sandy.addresses), "email addresses") | |
for addy in sandy.addresses: | |
print(addy.email_address, "belongs to", addy.user.name) | |
# get a single user, by ID | |
# this is useful for API urls where the user ID is in the URL | |
# e.g. /api/users/1 | |
user = session.get(User, 1) | |
print("Who is user ID=1?", user) | |
# select users matching names spongebob or sandy | |
# this is useful where you are searching for particular users | |
# and aren't sure how many there actually are | |
stmt = select(User).where(User.name.in_(["spongebob", "sandy"])) | |
# iterate over all rows returned in the above query | |
for user in session.scalars(stmt): | |
print("Matched user", user) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment