sqlalchemy_orm_quickstart_crud_operations_with_relationships.py

1from typing import List
2from typing import Optional
3from sqlalchemy import ForeignKey
4from sqlalchemy import String
5from sqlalchemy.orm import DeclarativeBase
6from sqlalchemy.orm import Mapped
7from sqlalchemy.orm import mapped_column
8from sqlalchemy.orm import relationship
9from sqlalchemy import create_engine
10from sqlalchemy.orm import Session
11
12class Base(DeclarativeBase):
13    pass
14
15class User(Base):
16    __tablename__ = "user_account"
17    id: Mapped[int] = mapped_column(primary_key=True)
18    name: Mapped[str] = mapped_column(String(30))
19    fullname: Mapped[Optional[str]]
20    addresses: Mapped[List["Address"]] = relationship(
21        back_populates="user", cascade="all, delete-orphan"
22    )
23    def __repr__(self) -> str:
24        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"
25
26class Address(Base):
27    __tablename__ = "address"
28    id: Mapped[int] = mapped_column(primary_key=True)
29    email_address: Mapped[str]
30    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
31    user: Mapped["User"] = relationship(back_populates="addresses")
32    def __repr__(self) -> str:
33        return f"Address(id={self.id!r}, email_address={self.email_address!r})"
34
35# Establish connectivity and create tables
36engine = create_engine("sqlite://", echo=True)
37Base.metadata.create_all(engine)
38
39# Create objects and persist them
40with Session(engine) as session:
41    spongebob = User(
42        name="spongebob",
43        fullname="Spongebob Squarepants",
44        addresses=[Address(email_address="spongebob@sqlalchemy.org")],
45    )
46    sandy = User(
47        name="sandy",
48        fullname="Sandy Cheeks",
49        addresses=[
50            Address(email_address="sandy@sqlalchemy.org"),
51            Address(email_address="sandy@google.com"),
52        ],
53    )
54    patrick = User(name="patrick", fullname="Patrick Star")
55    session.add_all([spongebob, sandy, patrick])
56    session.commit()
57
58# Simple SELECT
59with Session(engine) as session:
60    from sqlalchemy import select
61    stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))
62    for user in session.scalars(stmt):
63        print(user)
64
65# SELECT with JOIN
66with Session(engine) as session:
67    stmt = (
68        select(Address)
69        .join(Address.user)
70        .where(User.name == "sandy")
71        .where(Address.email_address == "sandy@sqlalchemy.org")
72    )
73    sandy_address = session.scalars(stmt).one()
74    print(sandy_address)
75
76# Make changes (Update)
77with Session(engine) as session:
78    stmt = select(User).where(User.name == "patrick")
79    patrick = session.scalars(stmt).one()
80    patrick.addresses.append(Address(email_address="patrickstar@sqlalchemy.org"))
81    session.commit()
82
83# Delete objects
84with Session(engine) as session:
85    sandy = session.get(User, 2)
86    session.delete(sandy)
87    session.commit()