Files
sqlalchemy-alembic/Essential-SQLAlchemy-2nd.org
T
2026-09-16 16:11:45 +05:00

27 KiB

Chapter 1: Schema and Types

SQLAlchemy v1

  import sqlalchemy
  print(sqlalchemy.__version__)
2.0.52
  from sqlalchemy import MetaData

  metadata = MetaData()
  from sqlalchemy import Table, Column, Integer, Numeric, String, CheckConstraint

  cookies = Table(
      "cookies",
      metadata,
      Column("cookie_id", Integer(), primary_key=True),
      Column("cookie_name", String(50), index=True),
      Column("cookie_recipie_url", String(255)),
      Column("cookie_sku", String(55)),
      Column("quantity", Integer()),
      Column("unit_cost", Numeric(12, 2)),
      CheckConstraint("unit_cost >= 0.00", name="unit_cost_positive"),
  )
  from datetime import datetime
  from sqlalchemy import DateTime, PrimaryKeyConstraint, UniqueConstraint

  users = Table(
      "users",
      metadata,
      Column("user_id", Integer()),
      Column("username", String(15), nullable=False),
      Column("email_address", String(255), nullable=False),
      Column("phone", String(20), nullable=False),
      Column("password", String(25), nullable=False),
      Column("created_on", DateTime(), default=datetime.now),
      Column("updated_on", DateTime(), default=datetime.now, onupdate=datetime.now),
      PrimaryKeyConstraint("user_id", name="user_pk"),
      UniqueConstraint("username", name="uix_username"),
  )
  @startuml
  skinparam linetype ortho
  skinparam nodesep 30
  skinparam ranksep 40

  skinparam entity {
  	BackgroundColor White
  	BorderColor Black
  	BorderThickness 1.2
  	FontStyle bold
  	FontSize 12
  }

  entity orders {
  	,**order_id** : PK
  	--
  	user_id : FK
  }
  entity line_items {
  	,**line_item_id**: PK
  	--
  	,**order_id** : FK
  	,**cookie_id** : FK
  	quantity
  	extended_cost
  }
  entity cookies {
  	,**cookie_id** : PK
  	--
  	cookie_name
  	cookie_reciepe_url
  	cookie_sku
  	quantity
  	unit_cost
  }
  entity users {
  	,**user_id** : PK
  	--
  	customer_number
  	username
  	email-address
  	phone
  	password
  	created_on
  	updated_on
  }

  orders ||--|{ line_items
  line_items }o--|| cookies
  orders }o--|| users

  orders -[hidden]right- line_items
  line_items -[hidden]right- cookies
  orders -[hidden]down- users
  @enduml

er.png

  from sqlalchemy import ForeignKey, Boolean

  orders = Table(
      "orders",
      metadata,
      Column("order_id", Integer(), primary_key=True),
      Column("user_id", ForeignKey("users.user_id")),
      Column("shipped", Boolean(), default=False),
  )


  line_items = Table(
      "line_items",
      metadata,
      Column("line_items_id", Integer(), primary_key=True),
      Column("order_id", ForeignKey("orders.order_id")),
      Column("cookie_id", ForeignKey("cookies.cookie_id")),
      Column("quantity", Integer()),
      Column("extended_cost", Numeric(12, 2)),
  )
  <<A1>>
  <<B1>>
  <<C1>>
  <<D1>>
  from sqlalchemy import create_engine

  engine = create_engine("sqlite:///:memory:", echo=True)
  metadata.create_all(engine)

SQLAlchemy v2

  #!/usr/bin/env python3
  from datetime import datetime
  from decimal import Decimal

  from sqlalchemy import (
      Boolean,
      CheckConstraint,
      DateTime,
      ForeignKey,
      Integer,
      Numeric,
      String,
      UniqueConstraint,
      create_engine,
  )
  from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


  class Base(DeclarativeBase):
      """Базовый класс для всех ORM-моделей."""


  class Cookie(Base):
      __tablename__ = "cookies"

      cookie_id: Mapped[int] = mapped_column(Integer, primary_key=True)
      cookie_name: Mapped[str | None] = mapped_column(String(50), index=True)
      cookie_recipie_url: Mapped[str | None] = mapped_column(String(255))
      cookie_sku: Mapped[str | None] = mapped_column(String(55))
      quantity: Mapped[int | None] = mapped_column(Integer)
      unit_cost: Mapped[Decimal | None] = mapped_column(Numeric(12, 2))

      __table_args__ = (CheckConstraint("unit_cost >= 0.00", name="unit_cost_positive"),)

      # Связь: одна cookie -> много line_items
      line_items: Mapped[list["LineItem"]] = relationship(back_populates="cookie")

      def __repr__(self) -> str:
          return f"Cookie(cookie_id={self.cookie_id!r}, cookie_name={self.cookie_name!r})"


  class User(Base):
      __tablename__ = "users"

      user_id: Mapped[int] = mapped_column(Integer, primary_key=True)
      username: Mapped[str] = mapped_column(String(15), nullable=False)
      email_address: Mapped[str] = mapped_column(String(255), nullable=False)
      phone: Mapped[str] = mapped_column(String(20), nullable=False)
      password: Mapped[str] = mapped_column(String(25), nullable=False)
      created_on: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.now)
      updated_on: Mapped[datetime | None] = mapped_column(
          DateTime, default=datetime.now, onupdate=datetime.now
      )

      __table_args__ = (UniqueConstraint("username", name="uix_username"),)

      # Связь: один user -> много orders
      orders: Mapped[list["Order"]] = relationship(back_populates="user")

      def __repr__(self) -> str:
          return f"User(user_id={self.user_id!r}, username={self.username!r})"


  class Order(Base):
      __tablename__ = "orders"

      order_id: Mapped[int] = mapped_column(Integer, primary_key=True)
      user_id: Mapped[int | None] = mapped_column(ForeignKey("users.user_id"))
      shipped: Mapped[bool | None] = mapped_column(Boolean, default=False)

      # Связи
      user: Mapped["User | None"] = relationship(back_populates="orders")
      line_items: Mapped[list["LineItem"]] = relationship(back_populates="order")

      def __repr__(self) -> str:
          return f"Order(order_id={self.order_id!r}, user_id={self.user_id!r})"


  class LineItem(Base):
      __tablename__ = "line_items"

      line_items_id: Mapped[int] = mapped_column(Integer, primary_key=True)
      order_id: Mapped[int | None] = mapped_column(ForeignKey("orders.order_id"))
      cookie_id: Mapped[int | None] = mapped_column(ForeignKey("cookies.cookie_id"))
      quantity: Mapped[int | None] = mapped_column(Integer)
      extended_cost: Mapped[Decimal | None] = mapped_column(Numeric(12, 2))

      # Связи
      order: Mapped["Order | None"] = relationship(back_populates="line_items")
      cookie: Mapped["Cookie | None"] = relationship(back_populates="line_items")

      def __repr__(self) -> str:
          return (
              f"LineItem(line_items_id={self.line_items_id!r}, "
              f"order_id={self.order_id!r}, cookie_id={self.cookie_id!r})"
          )


  if __name__ == "__main__":
      engine = create_engine("sqlite:///:memory:", echo=True)
      Base.metadata.create_all(engine)

Chapter 2: Working with Data via SQLAlchemy Core

  from datetime import datetime

  from sqlalchemy import (
      Boolean,
      CheckConstraint,
      Column,
      DateTime,
      ForeignKey,
      Integer,
      MetaData,
      Numeric,
      PrimaryKeyConstraint,
      String,
      Table,
      UniqueConstraint,
      create_engine,
  )
  from sqlparse import format as sqlformat


  def sqlf(
      sql: str | object,
      reindent=True,
      keyword_case="upper",
      indent_width=2,
      strip_comments=False,
      identifier_case="lower",
  ):
      if not isinstance(sql, str):
          sql = str(sql)
      return sqlformat(
          sql,
          encoding="utf8",
          reindent=reindent,
          keyword_case=keyword_case,
          indent_width=indent_width,
          strip_comments=strip_comments,
          identifier_case=identifier_case,
      )


  metadata = MetaData()

  cookies = Table(
      "cookies",
      metadata,
      Column("cookie_id", Integer(), primary_key=True),
      Column("cookie_name", String(50), index=True),
      Column("cookie_recipe_url", String(255), unique=True),
      Column("cookie_sku", String(55), unique=True),
      Column("quantity", Integer()),
      Column("unit_cost", Numeric(12, 2)),
      CheckConstraint("unit_cost >= 0.00", name="unit_cost_positive"),
  )

  users = Table(
      "users",
      metadata,
      Column("user_id", Integer()),
      Column("username", String(15), nullable=False),
      Column("email_address", String(255), nullable=False),
      Column("phone", String(20), nullable=False),
      Column("password", String(25), nullable=False),
      Column("created_on", DateTime(), default=datetime.now),
      Column("updated_on", DateTime(), default=datetime.now, onupdate=datetime.now),
      PrimaryKeyConstraint("user_id", name="user_pk"),
      UniqueConstraint("username", name="uix_username"),
  )

  orders = Table(
      "orders",
      metadata,
      Column("order_id", Integer(), primary_key=True),
      Column("user_id", ForeignKey("users.user_id")),
      Column("shipped", Boolean(), default=False),
  )


  line_items = Table(
      "line_items",
      metadata,
      Column("line_items_id", Integer(), primary_key=True),
      Column("order_id", ForeignKey("orders.order_id")),
      Column("cookie_id", ForeignKey("cookies.cookie_id")),
      Column("quantity", Integer()),
      Column("extended_cost", Numeric(12, 2)),
  )

  engine = create_engine("sqlite:///:memory:", echo=True)
  connection = engine.connect()
  metadata.create_all(engine)

SQLAlchemy v1

Insert Data

  ins = cookies.insert().values(
      cookie_name="chocolate chip",
      cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
      cookie_sku="CC01",
      quantity="12",
      unit_cost="0.50"
  )
  sqlf(ins)
INSERT INTO cookies (cookie_name, cookie_recipe_url, cookie_sku, quantity, unit_cost)
VALUES (:cookie_name, :cookie_recipe_url, :cookie_sku, :quantity, :unit_cost)
  import json

  json.dumps(ins.compile().params, indent=2)
{
  "cookie_name": "chocolate chip",
  "cookie_recipe_url": "http://some.aweso.me/cookie/recipe.html",
  "cookie_sku": "CC01",
  "quantity": "12",
  "unit_cost": "0.50"
}
  result = connection.execute(ins)
  result.inserted_primary_key
1
  ins = cookies.insert()
  result = connection.execute(
      ins,
      dict(
          cookie_name="dark chocolate chip",
          cookie_recipe_url="http://some.aweso.me/cookie/recipe_dark.html",
          cookie_sku="CC02",
          quantity=1,
          unit_cost=0.75,
      ),
  )
  result.inserted_primary_key
2
  from sqlalchemy import insert
  ins = insert(cookies).values(
      cookie_name="ligth chocolate chip",
      cookie_recipe_url="http://some.aweso.me/cookie/recipe_light.html",
      cookie_sku="CC03",
      quantity=10,
      unit_cost=0.5
  )
  result = connection.execute(ins)
  result.inserted_primary_key
3
  ins = cookies.insert()
  inventory_list = [
      {
          "cookie_name": "peanut butter",
          "cookie_recipe_url": "http://some.aweso.me/cookie/panut.html",
          "cookie_sku": "PB01",
          "quantity": 24,
          "unit_cost": 0.25
      },
      {
          "cookie_name": "oatmeal raisin",
          "cookie_recipe_url": "http://some.okay.me/cookie/raisin.html",
          "cookie_sku": "EWW01",
          "quantity": 100,
          "unit_cost": 1.00
      }
  ]

  result = connection.execute(ins, inventory_list)
  result.inserted_primary_key_rows
None
None

Querying Data

  from sqlalchemy.sql import select

  s = select(cookies)
  # ResultProxy
  rp = connection.execute(s)
  rp.fetchall()
1 chocolate chip http://some.aweso.me/cookie/recipe.html CC01 12 Decimal (0.50)
2 dark chocolate chip http://some.aweso.me/cookie/recipe_dark.html CC02 1 Decimal (0.75)
3 ligth chocolate chip http://some.aweso.me/cookie/recipe_light.html CC03 10 Decimal (0.50)
4 peanut butter http://some.aweso.me/cookie/panut.html PB01 24 Decimal (0.25)
5 oatmeal raisin http://some.okay.me/cookie/raisin.html EWW01 100 Decimal (1.00)
  s = cookies.select()
  rp = connection.execute(s)
  results = rp.fetchall()
  results
1 chocolate chip http://some.aweso.me/cookie/recipe.html CC01 12 Decimal (0.50)
2 dark chocolate chip http://some.aweso.me/cookie/recipe_dark.html CC02 1 Decimal (0.75)
3 ligth chocolate chip http://some.aweso.me/cookie/recipe_light.html CC03 10 Decimal (0.50)
4 peanut butter http://some.aweso.me/cookie/panut.html PB01 24 Decimal (0.25)
5 oatmeal raisin http://some.okay.me/cookie/raisin.html EWW01 100 Decimal (1.00)
ResultProxy
  first_row = results[0]
  (
      first_row[1],
      first_row.cookie_name,
      # так не работает в SQLAlchemy 2
      #first_row[cookies.c.cookie_name],
      # так работает в SQLAlchemy 2
      first_row._mapping[cookies.c.cookie_name]
  )
chocolate chip chocolate chip chocolate chip
  rp = connection.execute(s)
  for record in rp:
      print(record.cookie_name)
2026-09-16 15:38:27,551 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_id, cookies.cookie_name, cookies.cookie_recipe_url, cookies.cookie_sku, cookies.quantity, cookies.unit_cost 
FROM cookies
2026-09-16 15:38:27,551 INFO sqlalchemy.engine.Engine [cached since 0.09296s ago] ()
chocolate chip
dark chocolate chip
ligth chocolate chip
peanut butter
oatmeal raisin
Controlling the Columns in the Query
  s = select(cookies.c.cookie_name, cookies.c.quantity)
  rp = connection.execute(s)
  print(rp.keys())
  result = rp.first()
  print(result)
2026-09-16 15:38:27,565 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies
2026-09-16 15:38:27,565 INFO sqlalchemy.engine.Engine [generated in 0.00014s] ()
RMKeyView(['cookie_name', 'quantity'])
('chocolate chip', 12)
Ordering
  s = select(cookies.c.cookie_name, cookies.c.quantity)
  s = s.order_by(cookies.c.quantity)
  rp = connection.execute(s)
  for cookie in rp:
      print("{} - {}".format(cookie.quantity, cookie.cookie_name))
2026-09-16 15:38:27,579 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies ORDER BY cookies.quantity
2026-09-16 15:38:27,579 INFO sqlalchemy.engine.Engine [generated in 0.00012s] ()
1 - dark chocolate chip
10 - ligth chocolate chip
12 - chocolate chip
24 - peanut butter
100 - oatmeal raisin
  from sqlalchemy import desc
  s = select(cookies.c.cookie_name, cookies.c.quantity)
  s = s.order_by(desc(cookies.c.quantity))
  rp = connection.execute(s)
  rp.fetchall()
oatmeal raisin 100
peanut butter 24
chocolate chip 12
ligth chocolate chip 10
dark chocolate chip 1
Limiting
  s = select(cookies.c.cookie_name, cookies.c.quantity)
  s = s.order_by(cookies.c.quantity)
  s = s.limit(2)
  rp = connection.execute(s)
  print([result.cookie_name for result in rp])
2026-09-16 15:38:27,623 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies ORDER BY cookies.quantity
 LIMIT ? OFFSET ?
2026-09-16 15:38:27,623 INFO sqlalchemy.engine.Engine [generated in 0.00015s] (2, 0)
['dark chocolate chip', 'ligth chocolate chip']
Functions and Labels
  from sqlalchemy.sql import func

  s = select(func.sum(cookies.c.quantity))
  rp = connection.execute(s)
  print(rp.scalar())
2026-09-16 15:38:27,637 INFO sqlalchemy.engine.Engine SELECT sum(cookies.quantity) AS sum_1 
FROM cookies
2026-09-16 15:38:27,637 INFO sqlalchemy.engine.Engine [generated in 0.00015s] ()
147
  s = select(func.count(cookies.c.cookie_name))
  rp = connection.execute(s)
  record = rp.first()
  print(record._fields)
  print(record.count_1)
2026-09-16 15:38:27,650 INFO sqlalchemy.engine.Engine SELECT count(cookies.cookie_name) AS count_1 
FROM cookies
2026-09-16 15:38:27,650 INFO sqlalchemy.engine.Engine [generated in 0.00012s] ()
('count_1',)
5
  s = select(func.count(cookies.c.cookie_name).label('inventory_count'))
  rp = connection.execute(s)
  record = rp.first()
  print(record._fields)
  print(record.inventory_count)
2026-09-16 15:38:27,664 INFO sqlalchemy.engine.Engine SELECT count(cookies.cookie_name) AS inventory_count 
FROM cookies
2026-09-16 15:38:27,664 INFO sqlalchemy.engine.Engine [generated in 0.00016s] ()
('inventory_count',)
5
Filtering
  s = select(cookies).where(cookies.c.cookie_name == "chocolate chip")
  rp = connection.execute(s)
  record = rp.first()
  [
      (k, repr(v) if hasattr(v, "quantize") else v)
      for k, v in dict(record._mapping.items()).items()
  ]
cookie_id 1
cookie_name chocolate chip
cookie_recipe_url http://some.aweso.me/cookie/recipe.html
cookie_sku CC01
quantity 12
unit_cost Decimal('0.50')
  s = select(cookies).where(cookies.c.cookie_name.like("%chocolate%"))
  rp = connection.execute(s)
  for record in rp:
      print(record.cookie_name)
2026-09-16 15:38:27,710 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_id, cookies.cookie_name, cookies.cookie_recipe_url, cookies.cookie_sku, cookies.quantity, cookies.unit_cost 
FROM cookies 
WHERE cookies.cookie_name LIKE ?
2026-09-16 15:38:27,710 INFO sqlalchemy.engine.Engine [generated in 0.00015s] ('%chocolate%',)
chocolate chip
dark chocolate chip
ligth chocolate chip
  s = select(cookies.c.cookie_name, "SKU-" + cookies.c.cookie_sku)
  [row for row in connection.execute(s)]
chocolate chip SKU-CC01
dark chocolate chip SKU-CC02
ligth chocolate chip SKU-CC03
peanut butter SKU-PB01
oatmeal raisin SKU-EWW01
  from sqlalchemy import cast

  s = select(
      cookies.c.cookie_name,
      cast((cookies.c.quantity * cookies.c.unit_cost), Numeric(12, 2)).label("inv_cost"),
  )
  [(row.cookie_name, repr(row.inv_cost)) for row in connection.execute(s)]
chocolate chip Decimal('6.00')
dark chocolate chip Decimal('0.75')
ligth chocolate chip Decimal('5.00')
peanut butter Decimal('6.00')
oatmeal raisin Decimal('100.00')
Conjunctions
  from sqlalchemy import and_, or_, not_

  s = select(cookies).where(and_(cookies.c.quantity > 23, cookies.c.unit_cost < 0.40))
  [row.cookie_name for row in connection.execute(s)]
peanut butter
  s = select(cookies).where(
      or_(cookies.c.quantity.between(10, 50), cookies.c.cookie_name.contains("chip"))
  )
  [(row.cookie_name,) for row in connection.execute(s)]
chocolate chip
dark chocolate chip
ligth chocolate chip
peanut butter

Updating Data

  from sqlalchemy import update

  u = update(cookies).where(cookies.c.cookie_name == "chocolate chip")
  u = u.values(quantity=(cookies.c.quantity + 120))
  result = connection.execute(u)
  result.rowcount
1
  s = select(cookies).where(cookies.c.cookie_name == "chocolate chip")
  result = connection.execute(s).first()
  [
      (
          key,
          (
              repr(result._mapping[key])
              if hasattr(result._mapping[key], "quantize")
              else result._mapping[key]
          ),
      )
      for key in result._mapping.keys()
  ]
cookie_id 1
cookie_name chocolate chip
cookie_recipe_url http://some.aweso.me/cookie/recipe.html
cookie_sku CC01
quantity 132
unit_cost Decimal('0.50')

Deleting Data

  from sqlalchemy import delete

  u = delete(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
  result = connection.execute(u)
  result.rowcount
1
  s = select(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
  result = connection.execute(s).fetchall()
  len(result)
0

Joins

  customer_list = [
      {
          "username": "cookiemon",
          "email_address": "mon@cookie.com",
          "phone": "111-111-1111",
          "password": "password",
      },
      {
          "username": "cakeeater",
          "email_address": "cakeeater@cake.com",
          "phone": "222-222-2222",
          "password": "password",
      },
      {
          "username": "pieguy",
          "email_address": "guy@pie.com",
          "phone": "333-333-3333",
          "password": "password",
      },
  ]
  ins = users.insert()
  result = connection.execute(ins, customer_list)
  ins = insert(orders).values(user_id=1, order_id=1)
  result = connection.execute(ins)
  ins = insert(line_items)
  order_items = [
      {"order_id": 1, "cookie_id": 1, "quantity": 2, "extended_cost": 1.00},
      {"order_id": 1, "cookie_id": 3, "quantity": 12, "extended_cost": 3.00},
  ]
  result = connection.execute(ins, order_items)
  ins = insert(orders).values(user_id=2, order_id=2)
  result = connection.execute(ins)
  ins = insert(line_items)
  order_items = [
      {"order_id": 2, "cookie_id": 1, "quantity": 24, "extended_cost": 12.00},
      {"order_id": 2, "cookie_id": 4, "quantity": 6, "extended_cost": 6.00},
  ]
  result = connection.execute(ins, order_items)
  s = select(line_items)
  rp = connection.execute(s)
  [tuple(f for f in rp.keys())] + [row for row in rp]
line_items_id order_id cookie_id quantity extended_cost
1 1 1 2 Decimal (1.00)
2 1 3 12 Decimal (3.00)
3 2 1 24 Decimal (12.00)
4 2 4 6 Decimal (6.00)
  s = select(orders)
  rp = connection.execute(s)
  [tuple(f for f in rp.keys())] + [row for row in rp]
order_id user_id shipped
1 1 False
2 2 False
  s = select(users)
  rp = connection.execute(s)
  [tuple(f for f in rp.keys())] + [row for row in rp]
user_id username email_address phone password created_on updated_on
1 cookiemon mon@cookie.com 111-111-1111 password datetime.datetime (2026 9 16 15 39 54 867291) datetime.datetime (2026 9 16 15 39 54 867299)
2 cakeeater cakeeater@cake.com 222-222-2222 password datetime.datetime (2026 9 16 15 39 54 867301) datetime.datetime (2026 9 16 15 39 54 867302)
3 pieguy guy@pie.com 333-333-3333 password datetime.datetime (2026 9 16 15 39 54 867302) datetime.datetime (2026 9 16 15 39 54 867303)
  columns = [
      orders.c.order_id,
      users.c.username,
      users.c.phone,
      cookies.c.cookie_name,
      line_items.c.quantity,
      line_items.c.extended_cost,
  ]

  cookiemon_orders = (
      select(*columns)
      .select_from(
          orders
          .join(users, orders.c.user_id == users.c.user_id)
          .join(line_items, orders.c.order_id == line_items.c.order_id)
          .join(cookies, line_items.c.cookie_id == cookies.c.cookie_id)
      )
      .where(users.c.username == "cookiemon")
  )
  result = connection.execute(cookiemon_orders)
  [row for row in result]