Pavel Vavilin
127 KiB
127 KiB
- Chapter 1: Schema and Types
- Chapter 2: Working with Data via SQLAlchemy Core
- Chapter 3: Exceptions and Transactions
- Chapter 4: Testing
- Chapter 5: Reflection
- Chapter 6: Defining Schema with SQLAlchemy ORM
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| 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 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.rowcount1
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.rowcount1
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 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) |
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| 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,
),
)
connection.commit()
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)
connection.commit()
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)
connection.commit()
result.rowcount| 2 |
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 | chocolate chip | 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.rowcount1
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.rowcount1
s = select(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
result = connection.execute(s).all()
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)
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) |
Chapter 3: Exceptions and Transactions
SQLAlchemy v1
import sys
sys.stderr = sys.stdout
from datetime import datetime
from sqlalchemy import (
Boolean,
CheckConstraint,
Column,
DateTime,
ForeignKey,
Integer,
MetaData,
Numeric,
String,
Table,
create_engine,
PrimaryKeyConstraint,
UniqueConstraint,
)
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"),
CheckConstraint("quantity >= 0", name="quantity_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)AttributeError
import logging
from sqlalchemy import select, insert
ins = insert(users).values(
username="cookiemon",
email_address="mon@cookie.com",
phone="111-111-1111",
password="password",
)
result = connection.execute(ins)
s = select(users.c.username)
results = connection.execute(s)
try:
[(result.username, result.password) for result in results]
except AttributeError as exc:
logging.exception(results)2026-09-18 12:06:22,953 INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
2026-09-18 12:06:22,954 INFO sqlalchemy.engine.Engine INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)
INFO:sqlalchemy.engine.Engine:INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)
2026-09-18 12:06:22,954 INFO sqlalchemy.engine.Engine [generated in 0.00036s] ('cookiemon', 'mon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:06:22.953895', '2026-09-18 12:06:22.953904')
INFO:sqlalchemy.engine.Engine:[generated in 0.00036s] ('cookiemon', 'mon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:06:22.953895', '2026-09-18 12:06:22.953904')
2026-09-18 12:06:22,954 INFO sqlalchemy.engine.Engine SELECT users.username
FROM users
INFO:sqlalchemy.engine.Engine:SELECT users.username
FROM users
2026-09-18 12:06:22,954 INFO sqlalchemy.engine.Engine [generated in 0.00014s] ()
INFO:sqlalchemy.engine.Engine:[generated in 0.00014s] ()
ERROR:root:<sqlalchemy.engine.cursor.CursorResult object at 0x7fa8f6747950>
Traceback (most recent call last):
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/result.py", line 211, in _key_not_found
self._key_fallback(key, None)
~~~~~~~~~~~~~~~~~~^^^^^^^^^^^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/cursor.py", line 851, in _key_fallback
raise exc.NoSuchColumnError(
...<2 lines>...
) from err
sqlalchemy.exc.NoSuchColumnError: Could not locate column in row for column 'password'
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/tmp/babel-sg4Ps9/python-VQah8h", line 17, in <module>
[(result.username, result.password) for result in results]
^^^^^^^^^^^^^^^
File "lib/sqlalchemy/cyextension/resultproxy.pyx", line 66, in sqlalchemy.cyextension.resultproxy.BaseRow.__getattr__
File "lib/sqlalchemy/cyextension/resultproxy.pyx", line 63, in sqlalchemy.cyextension.resultproxy.BaseRow._get_by_key_impl
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/result.py", line 213, in _key_not_found
raise AttributeError(ke.args[0]) from ke
AttributeError: Could not locate column in row for column 'password'
IntegrityError
from sqlalchemy.exc import IntegrityError
s = select(users.c.username)
print(connection.execute(s).fetchall())
ins = insert(users).values(
username="cookiemon",
email_address="damon@cookie.com",
phone="111-111-1111",
password="password",
)
try:
result = connection.execute(ins)
except IntegrityError as exc:
logging.exception(str(ins))2026-09-18 12:06:22,969 INFO sqlalchemy.engine.Engine SELECT users.username
FROM users
INFO:sqlalchemy.engine.Engine:SELECT users.username
FROM users
2026-09-18 12:06:22,969 INFO sqlalchemy.engine.Engine [cached since 0.01439s ago] ()
INFO:sqlalchemy.engine.Engine:[cached since 0.01439s ago] ()
[('cookiemon',)]
2026-09-18 12:06:22,969 INFO sqlalchemy.engine.Engine INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)
INFO:sqlalchemy.engine.Engine:INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)
2026-09-18 12:06:22,969 INFO sqlalchemy.engine.Engine [cached since 0.0159s ago] ('cookiemon', 'damon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:06:22.969623', '2026-09-18 12:06:22.969627')
INFO:sqlalchemy.engine.Engine:[cached since 0.0159s ago] ('cookiemon', 'damon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:06:22.969623', '2026-09-18 12:06:22.969627')
ERROR:root:INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (:username, :email_address, :phone, :password, :created_on, :updated_on)
Traceback (most recent call last):
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlite3.IntegrityError: UNIQUE constraint failed: users.username
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/tmp/babel-sg4Ps9/python-Dqwfvm", line 13, in <module>
result = connection.execute(ins)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1421, in execute
return meth(
self,
distilled_parameters,
execution_options or NO_OPTIONS,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/sql/elements.py", line 526, in _execute_on_connection
return connection._execute_clauseelement(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
self, distilled_params, execution_options
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1643, in _execute_clauseelement
ret = self._execute_context(
dialect,
...<8 lines>...
cache_hit=cache_hit,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1848, in _execute_context
return self._exec_single_context(
~~~~~~~~~~~~~~~~~~~~~~~~~^
dialect, context, statement, parameters
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1988, in _exec_single_context
self._handle_dbapi_exception(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
e, str_statement, effective_parameters, cursor, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 2365, in _handle_dbapi_exception
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: users.username
[SQL: INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)]
[parameters: ('cookiemon', 'damon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:06:22.969623', '2026-09-18 12:06:22.969627')]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
Transactions
ins = cookies.insert()
inventory_list = [
{
"cookie_name": "chocolate chip",
"cookie_recipe_url": "http://some.aweso.me/cookie/recipe.html",
"cookie_sku": "CC01",
"quantity": "12",
"unit_cost": "0.5",
},
{
"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 = connection.execute(ins, inventory_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": 9,
"extended_cost": 4.50,
}
]
result = connection.execute(ins, order_items)
ins = insert(orders).values(user_id=1, order_id="2")
result = connection.execute(ins)
ins = insert(line_items)
order_items = [
{
"order_id": 2,
"cookie_id": 1,
"quantity": 4,
"extended_cost": 1.50,
},
{
"order_id": 2,
"cookie_id": 2,
"quantity": 1,
"extended_cost": 4.50,
},
]
result = connection.execute(ins, order_items) from sqlalchemy import update
def ship_it(order_id: int) -> None:
s = select(line_items.c.cookie_id, line_items.c.quantity)
s = s.where(line_items.c.order_id == order_id)
cookies_to_ship = connection.execute(s)
for cookie in cookies_to_ship:
u = (
update(cookies)
.where(cookies.c.cookie_id == cookie.cookie_id)
.values(quantity=cookies.c.quantity - cookie.quantity)
)
result = connection.execute(u)
u = update(orders).where(orders.c.order_id == order_id).values(shipped=True)
connection.execute(u)
print(f"Shipped order ID: {order_id}") ship_it(1)
s = select(cookies.c.cookie_name, cookies.c.quantity)
print(connection.execute(s).fetchall())2026-09-18 12:06:23,066 INFO sqlalchemy.engine.Engine SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
INFO:sqlalchemy.engine.Engine:SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
2026-09-18 12:06:23,066 INFO sqlalchemy.engine.Engine [generated in 0.00024s] (1,)
INFO:sqlalchemy.engine.Engine:[generated in 0.00024s] (1,)
2026-09-18 12:06:23,067 INFO sqlalchemy.engine.Engine UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
INFO:sqlalchemy.engine.Engine:UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
2026-09-18 12:06:23,067 INFO sqlalchemy.engine.Engine [generated in 0.00009s] (9, 1)
INFO:sqlalchemy.engine.Engine:[generated in 0.00009s] (9, 1)
2026-09-18 12:06:23,067 INFO sqlalchemy.engine.Engine UPDATE orders SET shipped=? WHERE orders.order_id = ?
INFO:sqlalchemy.engine.Engine:UPDATE orders SET shipped=? WHERE orders.order_id = ?
2026-09-18 12:06:23,067 INFO sqlalchemy.engine.Engine [generated in 0.00008s] (1, 1)
INFO:sqlalchemy.engine.Engine:[generated in 0.00008s] (1, 1)
Shipped order ID: 1
2026-09-18 12:06:23,067 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity
FROM cookies
INFO:sqlalchemy.engine.Engine:SELECT cookies.cookie_name, cookies.quantity
FROM cookies
2026-09-18 12:06:23,067 INFO sqlalchemy.engine.Engine [generated in 0.00006s] ()
INFO:sqlalchemy.engine.Engine:[generated in 0.00006s] ()
[('chocolate chip', 3), ('dark chocolate chip', 1)]
try:
ship_it(2)
except IntegrityError:
logging.exception("")2026-09-18 12:07:29,522 INFO sqlalchemy.engine.Engine SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
INFO:sqlalchemy.engine.Engine:SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
2026-09-18 12:07:29,522 INFO sqlalchemy.engine.Engine [cached since 66.46s ago] (2,)
INFO:sqlalchemy.engine.Engine:[cached since 66.46s ago] (2,)
2026-09-18 12:07:29,523 INFO sqlalchemy.engine.Engine UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
INFO:sqlalchemy.engine.Engine:UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
2026-09-18 12:07:29,523 INFO sqlalchemy.engine.Engine [cached since 66.46s ago] (4, 1)
INFO:sqlalchemy.engine.Engine:[cached since 66.46s ago] (4, 1)
ERROR:root:
Traceback (most recent call last):
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlite3.IntegrityError: CHECK constraint failed: quantity_positive
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/tmp/babel-sg4Ps9/python-swDwMp", line 2, in <module>
ship_it(2)
~~~~~~~^^^
File "<string>", line 14, in ship_it
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1421, in execute
return meth(
self,
distilled_parameters,
execution_options or NO_OPTIONS,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/sql/elements.py", line 526, in _execute_on_connection
return connection._execute_clauseelement(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
self, distilled_params, execution_options
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1643, in _execute_clauseelement
ret = self._execute_context(
dialect,
...<8 lines>...
cache_hit=cache_hit,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1848, in _execute_context
return self._exec_single_context(
~~~~~~~~~~~~~~~~~~~~~~~~~^
dialect, context, statement, parameters
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1988, in _exec_single_context
self._handle_dbapi_exception(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
e, str_statement, effective_parameters, cursor, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 2365, in _handle_dbapi_exception
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) CHECK constraint failed: quantity_positive
[SQL: UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?]
[parameters: (4, 1)]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
def ship_it(order_id: int) -> None:
s = select(line_items.c.cookie_id, line_items.c.quantity)
s = s.where(line_items.c.order_id == order_id)
transaction = connection.begin()
cookies_to_ship = connection.execute(s).fetchall()
try:
for cookie in cookies_to_ship:
u = update(cookies).where(cookies.c.cookie_id == cookie.cookie_id).values(quantity = cookies.c.quantity -cookie.quantity)
result = connection.execute(u)
u = update(orders).where(orders.c.order_id == order_id).values(shipped=True)
result = connection.execute(u)
print(f"Shipped order ID: {order_id}")
transaction.commit()
except IntegrityError as error:
transaction.rollback()
logging.exception()SQLAlchemy v2
#!/usr/bin/env python3
import sys
sys.stderr = sys.stdout
import logging
from datetime import datetime
from sqlalchemy import (
Boolean,
CheckConstraint,
Column,
DateTime,
ForeignKey,
Integer,
MetaData,
Numeric,
PrimaryKeyConstraint,
String,
Table,
UniqueConstraint,
create_engine,
insert,
select,
update,
)
from sqlalchemy.exc import IntegrityError
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"),
CheckConstraint("quantity >= 0", name="quantity_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)
with engine.begin() as connection:
metadata.create_all(connection)
# --- insert одного пользователя ---
ins = insert(users).values(
username="cookiemon",
email_address="mon@cookie.com",
phone="111-111-1111",
password="password",
)
connection.execute(ins)
# --- select возвращает Row, у которого есть .username ---
s = select(users.c.username, users.c.password)
results = connection.execute(s)
try:
[(row.username, row.password) for row in results]
except AttributeError:
logging.exception(results)
# --- проверка вставки ---
s = select(users.c.username)
print(connection.execute(s).fetchall())
# --- нарушение уникальности username ---
ins = insert(users).values(
username="cookiemon",
email_address="damon@cookie.com",
phone="111-111-1111",
password="password",
)
try:
with connection.begin_nested(): # SAVEPOINT, чтобы не убить внешнюю транзакцию
connection.execute(ins)
except IntegrityError:
logging.exception(str(ins))
# --- массовая вставка cookies ---
inventory_list = [
{
"cookie_name": "chocolate chip",
"cookie_recipe_url": "http://some.aweso.me/cookie/recipe.html",
"cookie_sku": "CC01",
"quantity": "12",
"unit_cost": "0.5",
},
{
"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.execute(insert(cookies), inventory_list)
# --- заказы ---
connection.execute(insert(orders).values(user_id=1, order_id="1"))
connection.execute(
insert(line_items),
[
{
"order_id": 1,
"cookie_id": 1,
"quantity": 9,
"extended_cost": 4.50,
}
],
)
connection.execute(insert(orders).values(user_id=1, order_id="2"))
connection.execute(
insert(line_items),
[
{
"order_id": 2,
"cookie_id": 1,
"quantity": 4,
"extended_cost": 1.50,
},
{
"order_id": 2,
"cookie_id": 2,
"quantity": 1,
"extended_cost": 4.50,
},
],
)
def ship_it(order_id: int) -> None:
"""Версия без явной транзакции — каждый execute коммитится сам."""
s = select(line_items.c.cookie_id, line_items.c.quantity).where(
line_items.c.order_id == order_id
)
cookies_to_ship = connection.execute(s).fetchall()
for cookie in cookies_to_ship:
u = (
update(cookies)
.where(cookies.c.cookie_id == cookie.cookie_id)
.values(quantity=cookies.c.quantity - cookie.quantity)
)
connection.execute(u)
u = update(orders).where(orders.c.order_id == order_id).values(shipped=True)
connection.execute(u)
print(f"Shipped order ID: {order_id}")
def ship_it_atomic(order_id: int) -> None:
"""Версия с явной транзакцией и rollback при ошибке."""
s = select(line_items.c.cookie_id, line_items.c.quantity).where(
line_items.c.order_id == order_id
)
with engine.begin() as conn:
cookies_to_ship = conn.execute(s).fetchall()
try:
for cookie in cookies_to_ship:
u = (
update(cookies)
.where(cookies.c.cookie_id == cookie.cookie_id)
.values(quantity=cookies.c.quantity - cookie.quantity)
)
conn.execute(u)
u = update(orders).where(orders.c.order_id == order_id).values(shipped=True)
conn.execute(u)
print(f"Shipped order ID: {order_id}")
except IntegrityError:
# with engine.begin() сам сделает rollback
logging.exception("Failed to ship order %s", order_id)
raise
# --- используем транзакционную версию ---
ship_it_atomic(1)
s = select(cookies.c.cookie_name, cookies.c.quantity)
with engine.connect() as conn:
print(conn.execute(s).fetchall())
try:
ship_it_atomic(2)
except IntegrityError:
logging.exception("")2026-09-18 12:21:58,909 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("cookies")
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("cookies")
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("users")
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("users")
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,910 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("orders")
2026-09-18 12:21:58,911 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,911 INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("orders")
2026-09-18 12:21:58,911 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,911 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("line_items")
2026-09-18 12:21:58,911 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,911 INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("line_items")
2026-09-18 12:21:58,911 INFO sqlalchemy.engine.Engine [raw sql] ()
2026-09-18 12:21:58,912 INFO sqlalchemy.engine.Engine
CREATE TABLE cookies (
cookie_id INTEGER NOT NULL,
cookie_name VARCHAR(50),
cookie_recipe_url VARCHAR(255),
cookie_sku VARCHAR(55),
quantity INTEGER,
unit_cost NUMERIC(12, 2),
PRIMARY KEY (cookie_id),
CONSTRAINT unit_cost_positive CHECK (unit_cost >= 0.00),
CONSTRAINT quantity_positive CHECK (quantity >= 0),
UNIQUE (cookie_recipe_url),
UNIQUE (cookie_sku)
)
2026-09-18 12:21:58,912 INFO sqlalchemy.engine.Engine [no key 0.00007s] ()
2026-09-18 12:21:58,912 INFO sqlalchemy.engine.Engine CREATE INDEX ix_cookies_cookie_name ON cookies (cookie_name)
2026-09-18 12:21:58,912 INFO sqlalchemy.engine.Engine [no key 0.00006s] ()
2026-09-18 12:21:58,913 INFO sqlalchemy.engine.Engine
CREATE TABLE users (
user_id INTEGER NOT NULL,
username VARCHAR(15) NOT NULL,
email_address VARCHAR(255) NOT NULL,
phone VARCHAR(20) NOT NULL,
password VARCHAR(25) NOT NULL,
created_on DATETIME,
updated_on DATETIME,
CONSTRAINT user_pk PRIMARY KEY (user_id),
CONSTRAINT uix_username UNIQUE (username)
)
2026-09-18 12:21:58,913 INFO sqlalchemy.engine.Engine [no key 0.00006s] ()
2026-09-18 12:21:58,913 INFO sqlalchemy.engine.Engine
CREATE TABLE orders (
order_id INTEGER NOT NULL,
user_id INTEGER,
shipped BOOLEAN,
PRIMARY KEY (order_id),
FOREIGN KEY(user_id) REFERENCES users (user_id)
)
2026-09-18 12:21:58,913 INFO sqlalchemy.engine.Engine [no key 0.00005s] ()
2026-09-18 12:21:58,913 INFO sqlalchemy.engine.Engine
CREATE TABLE line_items (
line_items_id INTEGER NOT NULL,
order_id INTEGER,
cookie_id INTEGER,
quantity INTEGER,
extended_cost NUMERIC(12, 2),
PRIMARY KEY (line_items_id),
FOREIGN KEY(order_id) REFERENCES orders (order_id),
FOREIGN KEY(cookie_id) REFERENCES cookies (cookie_id)
)
2026-09-18 12:21:58,913 INFO sqlalchemy.engine.Engine [no key 0.00005s] ()
2026-09-18 12:21:58,915 INFO sqlalchemy.engine.Engine INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)
2026-09-18 12:21:58,915 INFO sqlalchemy.engine.Engine [generated in 0.00019s] ('cookiemon', 'mon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:21:58.915122', '2026-09-18 12:21:58.915127')
2026-09-18 12:21:58,916 INFO sqlalchemy.engine.Engine SELECT users.username, users.password
FROM users
2026-09-18 12:21:58,916 INFO sqlalchemy.engine.Engine [generated in 0.00008s] ()
2026-09-18 12:21:58,916 INFO sqlalchemy.engine.Engine SELECT users.username
FROM users
2026-09-18 12:21:58,916 INFO sqlalchemy.engine.Engine [generated in 0.00007s] ()
[('cookiemon',)]
2026-09-18 12:21:58,916 INFO sqlalchemy.engine.Engine SAVEPOINT sa_savepoint_1
2026-09-18 12:21:58,916 INFO sqlalchemy.engine.Engine [no key 0.00006s] ()
2026-09-18 12:21:58,917 INFO sqlalchemy.engine.Engine INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)
2026-09-18 12:21:58,917 INFO sqlalchemy.engine.Engine [cached since 0.002053s ago] ('cookiemon', 'damon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:21:58.917037', '2026-09-18 12:21:58.917039')
2026-09-18 12:21:58,917 INFO sqlalchemy.engine.Engine ROLLBACK TO SAVEPOINT sa_savepoint_1
2026-09-18 12:21:58,917 INFO sqlalchemy.engine.Engine [no key 0.00006s] ()
ERROR:root:INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (:username, :email_address, :phone, :password, :created_on, :updated_on)
Traceback (most recent call last):
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlite3.IntegrityError: UNIQUE constraint failed: users.username
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/tmp/babel-sg4Ps9/python-ObJFq6", line 134, in <module>
connection.execute(ins)
~~~~~~~~~~~~~~~~~~^^^^^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1421, in execute
return meth(
self,
distilled_parameters,
execution_options or NO_OPTIONS,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/sql/elements.py", line 526, in _execute_on_connection
return connection._execute_clauseelement(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
self, distilled_params, execution_options
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1643, in _execute_clauseelement
ret = self._execute_context(
dialect,
...<8 lines>...
cache_hit=cache_hit,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1848, in _execute_context
return self._exec_single_context(
~~~~~~~~~~~~~~~~~~~~~~~~~^
dialect, context, statement, parameters
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1988, in _exec_single_context
self._handle_dbapi_exception(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
e, str_statement, effective_parameters, cursor, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 2365, in _handle_dbapi_exception
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: users.username
[SQL: INSERT INTO users (username, email_address, phone, password, created_on, updated_on) VALUES (?, ?, ?, ?, ?, ?)]
[parameters: ('cookiemon', 'damon@cookie.com', '111-111-1111', 'password', '2026-09-18 12:21:58.917037', '2026-09-18 12:21:58.917039')]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
2026-09-18 12:21:58,921 INFO sqlalchemy.engine.Engine INSERT INTO cookies (cookie_name, cookie_recipe_url, cookie_sku, quantity, unit_cost) VALUES (?, ?, ?, ?, ?)
INFO:sqlalchemy.engine.Engine:INSERT INTO cookies (cookie_name, cookie_recipe_url, cookie_sku, quantity, unit_cost) VALUES (?, ?, ?, ?, ?)
2026-09-18 12:21:58,921 INFO sqlalchemy.engine.Engine [generated in 0.00022s] [('chocolate chip', 'http://some.aweso.me/cookie/recipe.html', 'CC01', '12', 0.5), ('dark chocolate chip', 'http://some.aweso.me/cookie/recipe_dark.html', 'CC02', '1', 0.75)]
INFO:sqlalchemy.engine.Engine:[generated in 0.00022s] [('chocolate chip', 'http://some.aweso.me/cookie/recipe.html', 'CC01', '12', 0.5), ('dark chocolate chip', 'http://some.aweso.me/cookie/recipe_dark.html', 'CC02', '1', 0.75)]
2026-09-18 12:21:58,922 INFO sqlalchemy.engine.Engine INSERT INTO orders (order_id, user_id, shipped) VALUES (?, ?, ?)
INFO:sqlalchemy.engine.Engine:INSERT INTO orders (order_id, user_id, shipped) VALUES (?, ?, ?)
2026-09-18 12:21:58,922 INFO sqlalchemy.engine.Engine [generated in 0.00015s] ('1', 1, 0)
INFO:sqlalchemy.engine.Engine:[generated in 0.00015s] ('1', 1, 0)
2026-09-18 12:21:58,922 INFO sqlalchemy.engine.Engine INSERT INTO line_items (order_id, cookie_id, quantity, extended_cost) VALUES (?, ?, ?, ?)
INFO:sqlalchemy.engine.Engine:INSERT INTO line_items (order_id, cookie_id, quantity, extended_cost) VALUES (?, ?, ?, ?)
2026-09-18 12:21:58,922 INFO sqlalchemy.engine.Engine [generated in 0.00014s] (1, 1, 9, 4.5)
INFO:sqlalchemy.engine.Engine:[generated in 0.00014s] (1, 1, 9, 4.5)
2026-09-18 12:21:58,923 INFO sqlalchemy.engine.Engine INSERT INTO orders (order_id, user_id, shipped) VALUES (?, ?, ?)
INFO:sqlalchemy.engine.Engine:INSERT INTO orders (order_id, user_id, shipped) VALUES (?, ?, ?)
2026-09-18 12:21:58,923 INFO sqlalchemy.engine.Engine [cached since 0.001143s ago] ('2', 1, 0)
INFO:sqlalchemy.engine.Engine:[cached since 0.001143s ago] ('2', 1, 0)
2026-09-18 12:21:58,923 INFO sqlalchemy.engine.Engine INSERT INTO line_items (order_id, cookie_id, quantity, extended_cost) VALUES (?, ?, ?, ?)
INFO:sqlalchemy.engine.Engine:INSERT INTO line_items (order_id, cookie_id, quantity, extended_cost) VALUES (?, ?, ?, ?)
2026-09-18 12:21:58,923 INFO sqlalchemy.engine.Engine [generated in 0.00011s] [(2, 1, 4, 1.5), (2, 2, 1, 4.5)]
INFO:sqlalchemy.engine.Engine:[generated in 0.00011s] [(2, 1, 4, 1.5), (2, 2, 1, 4.5)]
2026-09-18 12:21:58,923 INFO sqlalchemy.engine.Engine COMMIT
INFO:sqlalchemy.engine.Engine:COMMIT
2026-09-18 12:21:58,924 INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
2026-09-18 12:21:58,924 INFO sqlalchemy.engine.Engine SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
INFO:sqlalchemy.engine.Engine:SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
2026-09-18 12:21:58,924 INFO sqlalchemy.engine.Engine [generated in 0.00010s] (1,)
INFO:sqlalchemy.engine.Engine:[generated in 0.00010s] (1,)
2026-09-18 12:21:58,926 INFO sqlalchemy.engine.Engine UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
INFO:sqlalchemy.engine.Engine:UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
2026-09-18 12:21:58,926 INFO sqlalchemy.engine.Engine [generated in 0.00009s] (9, 1)
INFO:sqlalchemy.engine.Engine:[generated in 0.00009s] (9, 1)
2026-09-18 12:21:58,926 INFO sqlalchemy.engine.Engine UPDATE orders SET shipped=? WHERE orders.order_id = ?
INFO:sqlalchemy.engine.Engine:UPDATE orders SET shipped=? WHERE orders.order_id = ?
2026-09-18 12:21:58,926 INFO sqlalchemy.engine.Engine [generated in 0.00012s] (1, 1)
INFO:sqlalchemy.engine.Engine:[generated in 0.00012s] (1, 1)
Shipped order ID: 1
2026-09-18 12:21:58,926 INFO sqlalchemy.engine.Engine COMMIT
INFO:sqlalchemy.engine.Engine:COMMIT
2026-09-18 12:21:58,927 INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
2026-09-18 12:21:58,927 INFO sqlalchemy.engine.Engine SELECT cookies.cookie_name, cookies.quantity
FROM cookies
INFO:sqlalchemy.engine.Engine:SELECT cookies.cookie_name, cookies.quantity
FROM cookies
2026-09-18 12:21:58,927 INFO sqlalchemy.engine.Engine [generated in 0.00012s] ()
INFO:sqlalchemy.engine.Engine:[generated in 0.00012s] ()
[('chocolate chip', 3), ('dark chocolate chip', 1)]
2026-09-18 12:21:58,927 INFO sqlalchemy.engine.Engine ROLLBACK
INFO:sqlalchemy.engine.Engine:ROLLBACK
2026-09-18 12:21:58,927 INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
2026-09-18 12:21:58,927 INFO sqlalchemy.engine.Engine SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
INFO:sqlalchemy.engine.Engine:SELECT line_items.cookie_id, line_items.quantity
FROM line_items
WHERE line_items.order_id = ?
2026-09-18 12:21:58,927 INFO sqlalchemy.engine.Engine [cached since 0.002892s ago] (2,)
INFO:sqlalchemy.engine.Engine:[cached since 0.002892s ago] (2,)
2026-09-18 12:21:58,928 INFO sqlalchemy.engine.Engine UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
INFO:sqlalchemy.engine.Engine:UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?
2026-09-18 12:21:58,928 INFO sqlalchemy.engine.Engine [cached since 0.002047s ago] (4, 1)
INFO:sqlalchemy.engine.Engine:[cached since 0.002047s ago] (4, 1)
ERROR:root:Failed to ship order 2
Traceback (most recent call last):
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlite3.IntegrityError: CHECK constraint failed: quantity_positive
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/tmp/babel-sg4Ps9/python-ObJFq6", line 224, in ship_it_atomic
conn.execute(u)
~~~~~~~~~~~~^^^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1421, in execute
return meth(
self,
distilled_parameters,
execution_options or NO_OPTIONS,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/sql/elements.py", line 526, in _execute_on_connection
return connection._execute_clauseelement(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
self, distilled_params, execution_options
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1643, in _execute_clauseelement
ret = self._execute_context(
dialect,
...<8 lines>...
cache_hit=cache_hit,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1848, in _execute_context
return self._exec_single_context(
~~~~~~~~~~~~~~~~~~~~~~~~~^
dialect, context, statement, parameters
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1988, in _exec_single_context
self._handle_dbapi_exception(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
e, str_statement, effective_parameters, cursor, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 2365, in _handle_dbapi_exception
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) CHECK constraint failed: quantity_positive
[SQL: UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?]
[parameters: (4, 1)]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
2026-09-18 12:21:58,930 INFO sqlalchemy.engine.Engine ROLLBACK
INFO:sqlalchemy.engine.Engine:ROLLBACK
ERROR:root:
Traceback (most recent call last):
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlite3.IntegrityError: CHECK constraint failed: quantity_positive
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/tmp/babel-sg4Ps9/python-ObJFq6", line 241, in <module>
ship_it_atomic(2)
~~~~~~~~~~~~~~^^^
File "/tmp/babel-sg4Ps9/python-ObJFq6", line 224, in ship_it_atomic
conn.execute(u)
~~~~~~~~~~~~^^^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1421, in execute
return meth(
self,
distilled_parameters,
execution_options or NO_OPTIONS,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/sql/elements.py", line 526, in _execute_on_connection
return connection._execute_clauseelement(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
self, distilled_params, execution_options
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1643, in _execute_clauseelement
ret = self._execute_context(
dialect,
...<8 lines>...
cache_hit=cache_hit,
)
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1848, in _execute_context
return self._exec_single_context(
~~~~~~~~~~~~~~~~~~~~~~~~~^
dialect, context, statement, parameters
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1988, in _exec_single_context
self._handle_dbapi_exception(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
e, str_statement, effective_parameters, cursor, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 2365, in _handle_dbapi_exception
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/base.py", line 1969, in _exec_single_context
self.dialect.do_execute(
~~~~~~~~~~~~~~~~~~~~~~~^
cursor, str_statement, effective_parameters, context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/octopus/study/Python/sqlalchemy-study/.venv/lib/python3.14/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) CHECK constraint failed: quantity_positive
[SQL: UPDATE cookies SET quantity=(cookies.quantity - ?) WHERE cookies.cookie_id = ?]
[parameters: (4, 1)]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
Chapter 4: Testing
from datetime import datetime
from sqlalchemy import (
MetaData,
Table,
Column,
Integer,
Numeric,
String,
DateTime,
ForeignKey,
Boolean,
create_engine,
)
from sqlalchemy.sql import insert
class DataAccessLayer:
connection = None
engine = None
conn_string = None
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)),
Column("cookie_sku", String(55)),
Column("quantity", Integer()),
Column("unit_cost", Numeric(12, 2)),
)
users = Table(
"users",
metadata,
Column("user_id", Integer(), primary_key=True),
Column("customer_number", Integer(), autoincrement=True),
Column("username", String(15), nullable=False, unique=True),
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),
)
orders = Table(
"orders",
metadata,
Column("order_id", Integer()),
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)),
)
def db_init(self, conn_string):
self.engine = create_engine(conn_string or self.conn_string)
self.metadata.create_all(self.engine)
self.connection = self.engine.connect()
dal = DataAccessLayer()
def prep_db():
ins = dal.cookies.insert()
dal.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",
),
)
inventory_list = [
{
"cookie_name": "peanut butter",
"cookie_recipe_url": "http://some.aweso.me/cookie/peanut.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",
},
]
dal.connection.execute(ins, inventory_list)
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 = dal.users.insert()
dal.connection.execute(ins, customer_list)
ins = insert(dal.orders).values(user_id=1, order_id="wlk001")
dal.connection.execute(ins)
ins = insert(dal.line_items)
order_items = [
{"order_id": "wlk001", "cookie_id": 1, "quantity": 2, "extended_cost": 1.00},
{"order_id": "wlk001", "cookie_id": 3, "quantity": 12, "extended_cost": 3.00},
]
dal.connection.execute(ins, order_items)
ins = insert(dal.orders).values(user_id=2, order_id="ol001")
dal.connection.execute(ins)
ins = insert(dal.line_items)
order_items = [
{"order_id": "ol001", "cookie_id": 1, "quantity": 24, "extended_cost": 12.00},
{"order_id": "ol001", "cookie_id": 4, "quantity": 6, "extended_cost": 6.00},
]
dal.connection.execute(ins, order_items) from db import dal
from sqlalchemy.sql import select
def get_orders_by_customer(
cust_name: str, shipped: bool | None = None, details: bool = False
):
columns = [dal.orders.c.order_id, dal.users.c.username, dal.users.c.phone]
joins = dal.users.join(dal.orders)
if details:
columns.extend(
[
dal.cookies.c.cookie_name,
dal.line_items.c.quantity,
dal.line_items.c.extended_cost,
]
)
joins = joins.join(dal.line_items).join(dal.cookies)
cust_orders = select(*columns)
cust_orders = cust_orders.select_from(joins).where(
dal.users.c.username == cust_name
)
if shipped is not None:
cust_orders = cust_orders.where(dal.orders.c.shipped == shipped)
return dal.connection.execute(cust_orders).fetchall()UnitTest
import unittest
from decimal import Decimal
from app import get_orders_by_customer
from db import dal, prep_db
class TestApp(unittest.TestCase):
cookie_orders = [("wlk001", "cookiemon", "111-111-1111")]
cookie_details = [
(
"wlk001",
"cookiemon",
"111-111-1111",
"dark chocolate chip",
2,
Decimal("1.00"),
),
("wlk001", "cookiemon", "111-111-1111", "oatmeal raisin", 12, Decimal("3.00")),
]
@classmethod
def setUpClass(cls):
dal.db_init("sqlite:///:memory:")
prep_db()
def test_orders_by_customer_blank(self):
"""
- значение cust_name пустая строка,
- shipped = None
"""
results = get_orders_by_customer("")
self.assertEqual(results, [])
def test_orders_by_customer_blank_shipped(self):
"""
- значение cust_name пустая строка,
- shipped = True
"""
results = get_orders_by_customer("", True)
self.assertEqual(results, [])
def test_orders_by_customer_blank_notshipped(self):
"""
- значение cust_name пустая строка,
- shipped = False
"""
results = get_orders_by_customer("", False)
self.assertEqual(results, [])
def test_orders_by_customer_blank_details(self):
results = get_orders_by_customer("", details=True)
self.assertEqual(results, [])
def test_orders_by_customer_blank_shipped_details(self):
results = get_orders_by_customer("", True, True)
self.assertEqual(results, [])
def test_orders_by_customer_blank_notshipped_details(self):
results = get_orders_by_customer("", False, True)
self.assertEqual(results, [])
def test_orders_by_customer(self):
expected_results = [("wlk001", "cookiemon", "111-111-1111")]
results = get_orders_by_customer("cookiemon")
self.assertEqual(results, expected_results)
def test_orders_by_customer_shipped_only(self):
results = get_orders_by_customer("cookiemon", True)
self.assertEqual(results, [])
def test_orders_by_customer_unshipped_only(self):
expected_results = [("wlk001", "cookiemon", "111-111-1111")]
results = get_orders_by_customer("cookiemon", False)
self.assertEqual(results, expected_results)
def test_orders_by_customer_with_details(self):
expected_results = [
(
"wlk001",
"cookiemon",
"111-111-1111",
"dark chocolate chip",
2,
Decimal("1.00"),
),
(
"wlk001",
"cookiemon",
"111-111-1111",
"oatmeal raisin",
12,
Decimal("3.00"),
),
]
results = get_orders_by_customer("cookiemon", details=True)
self.assertEqual(results, expected_results)
def test_orders_by_customer_shipped_only_with_details(self):
results = get_orders_by_customer("cookiemon", True, True)
self.assertEqual(results, [])
def test_orders_by_customer_unshipped_only_details(self):
expected_results = [
(
"wlk001",
"cookiemon",
"111-111-1111",
"dark chocolate chip",
2,
Decimal("1.00"),
),
(
"wlk001",
"cookiemon",
"111-111-1111",
"oatmeal raisin",
12,
Decimal("3.00"),
),
]
results = get_orders_by_customer("cookiemon", False, True)
self.assertEqual(results, expected_results) import unittest
from decimal import Decimal
from unittest import mock
from app import get_orders_by_customer
from db import dal, prep_db
class TestApp(unittest.TestCase):
cookie_orders = [("wlk001", "cookiemon", "111-111-1111")]
cookie_details = [
(
"wlk001",
"cookiemon",
"111-111-1111",
"dark chocolate chip",
2,
Decimal("1.00"),
),
("wlk001", "cookiemon", "111-111-1111", "oatmeal raisin", 12, Decimal("3.00")),
]
@mock.patch("app.select")
@mock.patch("app.dal.connection")
def test_orders_by_customer_blank(self, mock_conn, mock_select):
mock_select.return_value.select_from.return_value.where.return_value = None
mock_conn.execute.return_value.fetchall.return_value = []
results = get_orders_by_customer("")
self.assertEqual(results, [])
@mock.patch("app.dal.connection")
def test_orders_by_customer_blank_shipped(self, mock_conn):
mock_conn.execute.return_value.fetchall.return_value = []
results = get_orders_by_customer("", True)
self.assertEqual(results, [])
@mock.patch("app.dal.connection")
def test_orders_by_customer(self, mock_conn):
mock_conn.execute.return_value.fetchall.return_value = self.cookie_orders
results = get_orders_by_customer("cookiemon")
self.assertEqual(results, self.cookie_orders)pytest
from decimal import Decimal
import pytest
from app import get_orders_by_customer
from db import dal, prep_db
@pytest.fixture(autouse=True, scope="class")
def setup_db():
dal.db_init("sqlite:///:memory:")
prep_db()
yield
# здесь можно закрыть соединение, если нужно:
# dal.connection.close()
def test_orders_by_customer_blank():
"""
- значение cust_name пустая строка,
- shipped = None
"""
results = get_orders_by_customer("")
assert results == []
def test_orders_by_customer_blank_shipped():
"""
- значение cust_name пустая строка,
- shipped = True
"""
results = get_orders_by_customer("", True)
assert results == []
def test_orders_by_customer_blank_notshipped():
"""
- значение cust_name пустая строка,
- shipped = False
"""
results = get_orders_by_customer("", False)
assert results == []
def test_orders_by_customer_blank_details():
results = get_orders_by_customer("", details=True)
assert results == []
def test_orders_by_customer_blank_shipped_details():
results = get_orders_by_customer("", True, True)
assert results == []
def test_orders_by_customer_blank_notshipped_details():
results = get_orders_by_customer("", False, True)
assert results == []
def test_orders_by_customer():
expected_results = [("wlk001", "cookiemon", "111-111-1111")]
results = get_orders_by_customer("cookiemon")
assert results == expected_results
def test_orders_by_customer_shipped_only():
results = get_orders_by_customer("cookiemon", True)
assert results == []
def test_orders_by_customer_unshipped_only():
expected_results = [("wlk001", "cookiemon", "111-111-1111")]
results = get_orders_by_customer("cookiemon", False)
assert results == expected_results
def test_orders_by_customer_with_details():
expected_results = [
(
"wlk001",
"cookiemon",
"111-111-1111",
"dark chocolate chip",
2,
Decimal("1.00"),
),
(
"wlk001",
"cookiemon",
"111-111-1111",
"oatmeal raisin",
12,
Decimal("3.00"),
),
]
results = get_orders_by_customer("cookiemon", details=True)
assert results == expected_results
def test_orders_by_customer_shipped_only_with_details():
results = get_orders_by_customer("cookiemon", True, True)
assert results == []
def test_orders_by_customer_unshipped_only_details():
expected_results = [
(
"wlk001",
"cookiemon",
"111-111-1111",
"dark chocolate chip",
2,
Decimal("1.00"),
),
(
"wlk001",
"cookiemon",
"111-111-1111",
"oatmeal raisin",
12,
Decimal("3.00"),
),
]
results = get_orders_by_customer("cookiemon", False, True)
assert results == expected_results from decimal import Decimal
from unittest import mock
import pytest
from app import get_orders_by_customer
from db import dal, prep_db
# --- общие тестовые данные ---
@pytest.fixture
def cookie_orders():
return [("wlk001", "cookiemon", "111-111-1111")]
@pytest.fixture
def cookie_details():
return [
(
"wlk001",
"cookiemon",
"111-111-1111",
"dark chocolate chip",
2,
Decimal("1.00"),
),
("wlk001", "cookiemon", "111-111-1111", "oatmeal raisin", 12, Decimal("3.00")),
]
# --- тесты ---
@mock.patch("app.select")
@mock.patch("app.dal.connection")
def test_orders_by_customer_blank(mock_conn, mock_select):
mock_select.return_value.select_from.return_value.where.return_value = None
mock_conn.execute.return_value.fetchall.return_value = []
results = get_orders_by_customer("")
assert results == []
@mock.patch("app.dal.connection")
def test_orders_by_customer_blank_shipped(mock_conn):
mock_conn.execute.return_value.fetchall.return_value = []
results = get_orders_by_customer("", True)
assert results == []
@mock.patch("app.dal.connection")
def test_orders_by_customer(mock_conn, cookie_orders):
mock_conn.execute.return_value.fetchall.return_value = cookie_orders
results = get_orders_by_customer("cookiemon")
assert results == cookie_ordersChapter 5: Reflection
SQLAlchemy v2
from sqlalchemy import MetaData, create_engine
metadata = MetaData()
engine = create_engine("sqlite:///Chinook_Sqlite.sqlite") from sqlalchemy import Table
artist = Table("Artist", metadata, autoload_with=engine)
artist.columns.keys()| ArtistId | Name |
from sqlalchemy import select
with engine.begin() as connection:
s = select(artist).limit(10)
result = connection.execute(s).fetchall()
result| 1 | AC/DC |
| 2 | Accept |
| 3 | Aerosmith |
| 4 | Alanis Morissette |
| 5 | Alice In Chains |
| 6 | Antônio Carlos Jobim |
| 7 | Apocalyptica |
| 8 | Audioslave |
| 9 | BackBeat |
| 10 | Billy Cobham |
album = Table('Album', metadata, autoload_with=engine)
res = metadata.tables['Album']
repr(res)Table('Album', MetaData(), Column('AlbumId', INTEGER(), table=<Album>, primary_key=True, nullable=False), Column('Title', NVARCHAR(length=160), table=<Album>, nullable=False), Column('ArtistId', INTEGER(), ForeignKey('Artist.ArtistId'), table=<Album>, nullable=False), schema=None)
album.foreign_keys{ForeignKey('Artist.ArtistId')}
Reflecting a Whole Database
metadata.reflect(bind=engine)
list(metadata.tables.keys())| Artist | Album | Customer | Employee | Genre | Invoice | InvoiceLine | Track | MediaType | Playlist | PlaylistTrack |
Chapter 6: Defining Schema with SQLAlchemy ORM
SQLAlchemy v1
import re
import logging
from ruff_format import format_string
from sqlalchemy import Table, Column, Integer, Numeric, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class MyBase:
@classmethod
def display(cls: Base) -> str:
raw = repr(cls.__table__).replace(f"<{cls.__tablename__}>", f'"{cls.__tablename__}"')
raw = re.sub(
r"CallableColumnDefault\(<function datetime\.now at 0x[0-9a-f]+>\)",
"NOW",
raw,
)
try:
return format_string(raw)
except Exception:
logging.exception(raw)
class Cookie(Base, MyBase):
__tablename__ = "cookies"
cookie_id = Column(Integer(), primary_key=True)
cookie_name = Column(String(50), index=True)
cookie_recipe_url = Column(String(255))
cookie_sku = Column(String(55))
quantity = Column(Integer())
unit_cost = Column(Numeric(12, 2))
Cookie.display()Table(
"cookies",
MetaData(),
Column(
"cookie_id",
Integer(),
table="cookies",
primary_key=True,
nullable=False,
),
Column("cookie_name", String(length=50), table="cookies"),
Column("cookie_recipe_url", String(length=255), table="cookies"),
Column("cookie_sku", String(length=55), table="cookies"),
Column("quantity", Integer(), table="cookies"),
Column("unit_cost", Numeric(precision=12, scale=2), table="cookies"),
schema=None,
)
from datetime import datetime
from sqlalchemy import DateTime
class User(Base, MyBase):
__tablename__ = "users"
user_id = Column(Integer(), primary_key=True)
username = Column(String(12), nullable=False, unique=True)
email_address = Column(String(255), nullable=False)
phone = Column(String(20), nullable=False)
password = Column(String(25), nullable=False)
created_on = Column(DateTime(), default=datetime.now)
updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)
User.display()Table(
"users",
MetaData(),
Column(
"user_id", Integer(), table="users", primary_key=True, nullable=False
),
Column("username", String(length=12), table="users", nullable=False),
Column("email_address", String(length=255), table="users", nullable=False),
Column("phone", String(length=20), table="users", nullable=False),
Column("password", String(length=25), table="users", nullable=False),
Column("created_on", DateTime(), table="users", default=NOW),
Column("updated_on", DateTime(), table="users", onupdate=NOW, default=NOW),
schema=None,
)
Keys, Constraints, and Indexes
class SomeDataClass(Base):
__tablename__ = "somedataclass"
__table_args__ = (
ForeignKeyConstraint("id", "other_table.id"),
CheckConstraint("unit_cost >= 0.00", name="unit_cost_positive"),
)Relationships
from sqlalchemy import ForeignKey, Boolean
from sqlalchemy.orm import relationship, backref
class Order(Base, MyBase):
__tablename__ = "orders"
order_id = Column(Integer(), primary_key=True)
user_id = Column(Integer(), ForeignKey("users.user_id"))
shipped = Column(Boolean(), default=False)
user = relationship("User", backref=backref("orders", order_by=order_id))
Order.display()Table(
"orders",
MetaData(),
Column(
"order_id", Integer(), table="orders", primary_key=True, nullable=False
),
Column("user_id", Integer(), ForeignKey("users.user_id"), table="orders"),
Column(
"shipped",
Boolean(),
table="orders",
default=ScalarElementColumnDefault(False),
),
schema=None,
)
One-to-One
class LineItem(Base, MyBase):
__tablename__ = "line_items"
line_item_id = Column(Integer(), primary_key=True)
order_id = Column(Integer(), ForeignKey("orders.order_id"))
cookie_id = Column(Integer(), ForeignKey("cookies.cookie_id"))
quantity = Column(Integer())
extended_cost = Column(Numeric(12, 2))
order = relationship("Order", backref=backref("line_items", order_by=line_item_id))
cookie = relationship("Cookie", uselist=False)
LineItem.display()Table(
"line_items",
MetaData(),
Column(
"line_item_id",
Integer(),
table="line_items",
primary_key=True,
nullable=False,
),
Column(
"order_id", Integer(), ForeignKey("orders.order_id"), table="line_items"
),
Column(
"cookie_id",
Integer(),
ForeignKey("cookies.cookie_id"),
table="line_items",
),
Column("quantity", Integer(), table="line_items"),
Column("extended_cost", Numeric(precision=12, scale=2), table="line_items"),
schema=None,
)
Order.display()Table(
"orders",
MetaData(),
Column(
"order_id", Integer(), table="orders", primary_key=True, nullable=False
),
Column("user_id", Integer(), ForeignKey("users.user_id"), table="orders"),
Column(
"shipped",
Boolean(),
table="orders",
default=ScalarElementColumnDefault(False),
),
schema=None,
)
from sqlalchemy import create_engine
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)SQLAlchemy v2
import re
import logging
from datetime import datetime
from decimal import Decimal
from ruff_format import format_string
from sqlalchemy import (
Boolean,
DateTime,
ForeignKey,
Integer,
Numeric,
String,
create_engine,
select,
)
from sqlalchemy.orm import (
DeclarativeBase,
Mapped,
mapped_column,
relationship,
Session,
)
class Base(DeclarativeBase):
"""Базовый класс для всех декларативных моделей."""
class MyBase:
"""Миксин для красивого отображения таблицы."""
@classmethod
def display(cls) -> str:
raw = repr(cls.__table__).replace(
f"<{cls.__tablename__}>", f'"{cls.__tablename__}"'
)
raw = re.sub(
r"CallableColumnDefault\(<function datetime\.now at 0x[0-9a-f]+>\)",
"NOW",
raw,
)
try:
return format_string(raw)
except Exception:
logging.exception(raw)
return raw
class Cookie(Base, MyBase):
__tablename__ = "cookies"
cookie_id: Mapped[int] = mapped_column(Integer, primary_key=True)
cookie_name: Mapped[str | None] = mapped_column(String(50), index=True)
cookie_recipe_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))
Cookie.display()Table(
"cookies",
MetaData(),
Column(
"cookie_id",
Integer(),
table="cookies",
primary_key=True,
nullable=False,
),
Column("cookie_name", String(length=50), table="cookies"),
Column("cookie_recipe_url", String(length=255), table="cookies"),
Column("cookie_sku", String(length=55), table="cookies"),
Column("quantity", Integer(), table="cookies"),
Column("unit_cost", Numeric(precision=12, scale=2), table="cookies"),
schema=None,
)
class User(Base, MyBase):
__tablename__ = "users"
user_id: Mapped[int] = mapped_column(Integer, primary_key=True)
username: Mapped[str] = mapped_column(String(12), nullable=False, unique=True)
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] = mapped_column(DateTime, default=datetime.now)
updated_on: Mapped[datetime] = mapped_column(
DateTime, default=datetime.now, onupdate=datetime.now
)
orders: Mapped[list["Order"]] = relationship(
back_populates="user", order_by="Order.order_id"
)
User.display()Table(
"users",
MetaData(),
Column(
"user_id", Integer(), table="users", primary_key=True, nullable=False
),
Column("username", String(length=12), table="users", nullable=False),
Column("email_address", String(length=255), table="users", nullable=False),
Column("phone", String(length=20), table="users", nullable=False),
Column("password", String(length=25), table="users", nullable=False),
Column(
"created_on", DateTime(), table="users", nullable=False, default=NOW
),
Column(
"updated_on",
DateTime(),
table="users",
nullable=False,
onupdate=NOW,
default=NOW,
),
schema=None,
)
Keys, Constraints, and Indexes
from sqlalchemy import CheckConstraint, ForeignKeyConstraint
class SomeDataClass(Base):
__tablename__ = "somedataclass"
__table_args__ = (
ForeignKeyConstraint(["id"], ["other_table.id"]),
CheckConstraint("unit_cost >= 0.00", name="unit_cost_positive"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
unit_cost: Mapped[Decimal | None] = mapped_column(Numeric(12, 2))Relationships
class Order(Base, MyBase):
__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", order_by="LineItem.line_item_id"
)
Order.display()Table(
"orders",
MetaData(),
Column(
"order_id", Integer(), table="orders", primary_key=True, nullable=False
),
Column("user_id", Integer(), ForeignKey("users.user_id"), table="orders"),
Column(
"shipped",
Boolean(),
table="orders",
default=ScalarElementColumnDefault(False),
),
schema=None,
)
One-to-One
class LineItem(Base, MyBase):
__tablename__ = "line_items"
line_item_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(uselist=False)
LineItem.display()Table(
"line_items",
MetaData(),
Column(
"line_item_id",
Integer(),
table="line_items",
primary_key=True,
nullable=False,
),
Column(
"order_id", Integer(), ForeignKey("orders.order_id"), table="line_items"
),
Column(
"cookie_id",
Integer(),
ForeignKey("cookies.cookie_id"),
table="line_items",
),
Column("quantity", Integer(), table="line_items"),
Column("extended_cost", Numeric(precision=12, scale=2), table="line_items"),
schema=None,
)
Order.display()Table(
"orders",
MetaData(),
Column(
"order_id", Integer(), table="orders", primary_key=True, nullable=False
),
Column("user_id", Integer(), ForeignKey("users.user_id"), table="orders"),
Column(
"shipped",
Boolean(),
table="orders",
default=ScalarElementColumnDefault(False),
),
schema=None,
)
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add(
Cookie(
cookie_name="chocolate chip",
cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
cookie_sku="CC01",
quantity=12,
unit_cost=Decimal("0.50"),
)
)
session.commit()
stmt = select(Cookie).where(Cookie.cookie_name == "chocolate chip")
for cookie in session.scalars(stmt):
print(cookie.cookie_name, cookie.unit_cost)