SQLAlchemy Quickstart Annotated

SQLAlchemy-Quickstart.py

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)