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