sqlalchemy-alembic/Essential-SQLAlchemy-2nd.org at b4907ae96c64f874477d1fc28dbbbda5fdbfab76

pimiento/sqlalchemy-alembic

Fork 0

Files

T

Pavel Vavilin b4907ae96c Essential SQLAlchemy: Chapter 2 (v2)

2026-09-16 19:36:54 +05:00

47 KiB

Raw Blame History

Chapter 1: Schema and Types
- SQLAlchemy v1
- SQLAlchemy v2
Chapter 2: Working with Data via SQLAlchemy Core
- SQLAlchemy v1
- SQLAlchemy v2

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

  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

SQLAlchemy v1

  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)

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

  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

  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

  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

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

  rp = connection.execute(s)
  for record in rp:
      print(record.cookie_name)

2026-09-16 18:20:06,397 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 18:20:06,398 INFO sqlalchemy.engine.Engine [cached since 0.1297s 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 18:20:06,420 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies
2026-09-16 18:20:06,421 INFO sqlalchemy.engine.Engine [generated in 0.00045s] ()
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 18:20:06,443 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies ORDER BY cookies.quantity
2026-09-16 18:20:06,443 INFO sqlalchemy.engine.Engine [generated in 0.00017s] ()
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 18:20:06,508 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies ORDER BY cookies.quantity
 LIMIT ? OFFSET ?
2026-09-16 18:20:06,512 INFO sqlalchemy.engine.Engine [generated in 0.00363s] (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 18:20:06,530 INFO sqlalchemy.engine.Engine SELECT sum(cookies.quantity) AS sum_1 
FROM cookies
2026-09-16 18:20:06,531 INFO sqlalchemy.engine.Engine [generated in 0.00018s] ()
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 18:20:06,548 INFO sqlalchemy.engine.Engine SELECT count(cookies.cookie_name) AS count_1 
FROM cookies
2026-09-16 18:20:06,548 INFO sqlalchemy.engine.Engine [generated in 0.00028s] ()
('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 18:20:06,572 INFO sqlalchemy.engine.Engine SELECT count(cookies.cookie_name) AS inventory_count 
FROM cookies
2026-09-16 18:20:06,572 INFO sqlalchemy.engine.Engine [generated in 0.00027s] ()
('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 18:20:06,635 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 18:20:06,638 INFO sqlalchemy.engine.Engine [generated in 0.00392s] ('%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

  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

  s = select(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
  result = connection.execute(s).fetchall()
  len(result)

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 18 20 6 947339)	datetime.datetime	(2026 9 16 18 20 6 947349)
2	cakeeater	cakeeater@cake.com	222-222-2222	password	datetime.datetime	(2026 9 16 18 20 6 947352)	datetime.datetime	(2026 9 16 18 20 6 947355)
3	pieguy	guy@pie.com	333-333-3333	password	datetime.datetime	(2026 9 16 18 20 6 947359)	datetime.datetime	(2026 9 16 18 20 6 947360)

  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]

1	cookiemon	111-111-1111	chocolate chip	2	Decimal	(1.00)
1	cookiemon	111-111-1111	ligth chocolate chip	12	Decimal	(3.00)

TODO SQLAlchemy v2

  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)),
  )

  from sqlalchemy.pool import StaticPool

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

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)
  connection.commit()
  result.inserted_primary_key

  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,
      ),
  )
  connection.commit()
  result.inserted_primary_key

  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)
  connection.commit()
  result.inserted_primary_key

  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)
  connection.commit()
  result.rowcount

Querying Data

  from sqlalchemy import select

  s = select(cookies)
  result = connection.execute(s)
  result.all()

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 = select(cookies)
  results = connection.execute(s).all()
  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)

Result

  first_row = results[0]
  (
      first_row[1],
      first_row.cookie_name,
      first_row._mapping[cookies.c.cookie_name]
  )

chocolate chip

  rp = connection.execute(select(cookies))
  for record in rp:
      print(record.cookie_name)

2026-09-16 19:25:14,506 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 19:25:14,506 INFO sqlalchemy.engine.Engine [cached since 0.09276s 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 19:25:14,520 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies
2026-09-16 19:25:14,521 INFO sqlalchemy.engine.Engine [generated in 0.00015s] ()
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 19:25:14,532 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies ORDER BY cookies.quantity
2026-09-16 19:25:14,532 INFO sqlalchemy.engine.Engine [generated in 0.00017s] ()
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.all()

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 19:25:14,582 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity 
FROM cookies ORDER BY cookies.quantity
 LIMIT ? OFFSET ?
2026-09-16 19:25:14,582 INFO sqlalchemy.engine.Engine [generated in 0.00020s] (2, 0)
['dark chocolate chip', 'ligth chocolate chip']

Functions and Labels

  from sqlalchemy import func

  s = select(func.sum(cookies.c.quantity))
  rp = connection.execute(s)
  print(rp.scalar())

2026-09-16 19:25:14,596 INFO sqlalchemy.engine.Engine SELECT sum(cookies.quantity) AS sum_1 
FROM cookies
2026-09-16 19:25:14,596 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 19:25:14,609 INFO sqlalchemy.engine.Engine SELECT count(cookies.cookie_name) AS count_1 
FROM cookies
2026-09-16 19:25:14,610 INFO sqlalchemy.engine.Engine [generated in 0.00014s] ()
('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 19:25:14,623 INFO sqlalchemy.engine.Engine SELECT count(cookies.cookie_name) AS inventory_count 
FROM cookies
2026-09-16 19:25:14,623 INFO sqlalchemy.engine.Engine [generated in 0.00015s] ()
('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 19:25:14,672 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 19:25:14,672 INFO sqlalchemy.engine.Engine [generated in 0.00017s] ('%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_

  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)
  connection.commit()
  result.rowcount

  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)
  connection.commit()
  result.rowcount

  s = select(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
  result = connection.execute(s).all()
  len(result)

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)
  connection.commit()

  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)
  connection.commit()

  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 19 25 14 928945)	datetime.datetime	(2026 9 16 19 25 14 928953)
2	cakeeater	cakeeater@cake.com	222-222-2222	password	datetime.datetime	(2026 9 16 19 25 14 928955)	datetime.datetime	(2026 9 16 19 25 14 928956)
3	pieguy	guy@pie.com	333-333-3333	password	datetime.datetime	(2026 9 16 19 25 14 928957)	datetime.datetime	(2026 9 16 19 25 14 928958)

  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]

1	cookiemon	111-111-1111	chocolate chip	2	Decimal	(1.00)
1	cookiemon	111-111-1111	ligth chocolate chip	12	Decimal	(3.00)

47 KiB Raw Blame History

Chapter 1: Schema and Types

SQLAlchemy v1

SQLAlchemy v2

Chapter 2: Working with Data via SQLAlchemy Core

SQLAlchemy v1

Insert Data

Querying Data

ResultProxy

Controlling the Columns in the Query

Ordering

Limiting

Functions and Labels

Filtering

Conjunctions

Updating Data

Deleting Data

Joins

TODO SQLAlchemy v2

Insert Data

Querying Data

Result

Controlling the Columns in the Query

Ordering

Limiting

Functions and Labels

Filtering

Conjunctions

Updating Data

Deleting Data

Joins

47 KiB

Raw Blame History