Advanced queries and async sessions

from sqlalchemy import delete, insert, update

with SessionLocal.begin() as session:
    new_user = session.scalar(
        insert(User).values(name="Ada", email="ada@example.com")
        .returning(User)
    )
    changed = session.execute(
        update(User).where(User.is_active.is_(False))
        .values(is_active=True).returning(User.id, User.email)
    )
    session.execute(delete(User).where(User.created_at < cutoff))

INSERT INTO user_account (name, email, is_active)
VALUES ('Ada', 'ada@example.com', true) RETURNING *;

UPDATE user_account SET is_active = true
WHERE is_active = false RETURNING id, email;

DELETE FROM user_account WHERE created_at < :cutoff;

from sqlalchemy.dialects.postgresql import insert as pg_insert

session.execute(insert(User), [
    {"name": "Ada", "email": "ada@example.com"},
    {"name": "Lin", "email": "lin@example.com"},
])

stmt = pg_insert(User).values(name="Ada", email="ada@example.com")
stmt = stmt.on_conflict_do_update(
    index_elements=[User.email],
    set_={"name": stmt.excluded.name},
).returning(User.id, User.name)
row = session.execute(stmt).one()

INSERT INTO user_account (name, email) VALUES
  ('Ada', 'ada@example.com'), ('Lin', 'lin@example.com');

INSERT INTO user_account (name, email)
VALUES ('Ada', 'ada@example.com')
ON CONFLICT (email) DO UPDATE SET name = EXCLUDED.name
RETURNING id, name;

from sqlalchemy import case, cast, func, literal, String

count_addresses = func.count(Address.id)
status = case(
    (User.is_active, literal("active")),
    else_=literal("disabled"),
).label("status")

stmt = (
    select(User.id, func.upper(User.name).label("name_upper"),
           count_addresses.label("address_count"), status)
    .outerjoin(User.addresses)
    .group_by(User.id, User.name, User.is_active)
    .having(count_addresses >= 2)
)
typed = select(cast(User.id, String).label("id_as_text"))

SELECT u.id, upper(u.name) AS name_upper,
       count(a.id) AS address_count,
       CASE WHEN u.is_active THEN 'active' ELSE 'disabled' END AS status
FROM user_account u LEFT JOIN address a ON u.id = a.user_id
GROUP BY u.id, u.name, u.is_active
HAVING count(a.id) >= 2;

SELECT CAST(id AS VARCHAR) AS id_as_text FROM user_account;

from sqlalchemy import exists, union_all

counts = (
    select(Address.user_id, func.count().label("n"))
    .group_by(Address.user_id).subquery()
)
with_two = select(User.name, counts.c.n).join(
    counts, counts.c.user_id == User.id
).where(counts.c.n >= 2)

active_users = select(User.id, User.name).where(User.is_active).cte("active_users")
has_address = select(User.name).where(
    exists(select(Address.id).where(Address.user_id == User.id))
)
names = union_all(select(User.name.label("value")),
                  select(Address.email.label("value")))
rank = func.row_number().over(
    partition_by=Address.user_id, order_by=Address.id.desc()
).label("rank")

SELECT u.name, c.n FROM user_account u JOIN
  (SELECT user_id, count(*) n FROM address GROUP BY user_id) c
ON c.user_id = u.id WHERE c.n >= 2;

WITH active_users AS (
  SELECT id, name FROM user_account WHERE is_active = true
) SELECT * FROM active_users;

SELECT name FROM user_account u WHERE EXISTS
  (SELECT 1 FROM address a WHERE a.user_id = u.id);

SELECT name AS value FROM user_account
UNION ALL SELECT email AS value FROM address;

SELECT row_number() OVER (PARTITION BY user_id ORDER BY id DESC) FROM address;

from sqlalchemy.orm import aliased

address_count = (select(func.count(Address.id))
                 .where(Address.user_id == User.id)
                 .correlate(User).scalar_subquery())
stmt = select(User.name, address_count.label("address_count"))

ranked = (select(Address,
                 func.row_number().over(
                     partition_by=Address.user_id,
                     order_by=Address.id.desc()).label("rn"))
          .subquery())
newest = aliased(Address, ranked)
stmt = select(newest).where(ranked.c.rn == 1)
latest_addresses = session.scalars(stmt).all()

SELECT u.name, (SELECT count(*) FROM address AS a
                WHERE a.user_id = u.id) AS address_count
FROM user_account AS u;

SELECT r.id, r.user_id, r.email FROM (
  SELECT a.*, row_number() OVER
    (PARTITION BY a.user_id ORDER BY a.id DESC) AS rn
  FROM address AS a
) AS r WHERE r.rn = 1;

from sqlalchemy import literal, true

descendants = (select(Node.id, Node.parent_id, literal(0).label("depth"))
               .where(Node.id == root_id).cte("descendants", recursive=True))
n = Node.__table__.alias("n")
descendants = descendants.union_all(
    select(n.c.id, n.c.parent_id, descendants.c.depth + 1)
    .join(descendants, n.c.parent_id == descendants.c.id)
    .where(descendants.c.depth < 20)
)
stmt = select(descendants.c.id).select_from(descendants)

newest = (select(Address.email)
          .where(Address.user_id == User.id)
          .order_by(Address.id.desc()).limit(1).lateral("newest"))
stmt = select(User.name, newest.c.email).join(newest, true())

WITH RECURSIVE descendants(id, parent_id, depth) AS (
  SELECT id, parent_id, 0 FROM node WHERE id = :root_id
  UNION ALL
  SELECT n.id, n.parent_id, d.depth + 1 FROM node AS n
  JOIN descendants AS d ON n.parent_id = d.id WHERE d.depth < 20
) SELECT id FROM descendants;

SELECT u.name, newest.email FROM user_account AS u
JOIN LATERAL (
  SELECT a.email FROM address AS a WHERE a.user_id = u.id
  ORDER BY a.id DESC LIMIT 1
) AS newest ON true;

from sqlalchemy import CheckConstraint, Computed, Index, UniqueConstraint

class Product(Base):
    __tablename__ = "product"
    __table_args__ = (
        UniqueConstraint("tenant_id", "sku", name="uq_product_tenant_sku"),
        CheckConstraint("price_cents >= 0", name="ck_product_nonnegative_price"),
        Index("ix_product_tenant_active", "tenant_id", "is_active"),
    )
    id: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int]
    sku: Mapped[str] = mapped_column(String(64))
    price_cents: Mapped[int]
    quantity: Mapped[int]
    line_total: Mapped[int] = mapped_column(Computed("price_cents * quantity"))
    is_active: Mapped[bool] = mapped_column(default=True)

CREATE TABLE product (
  id SERIAL PRIMARY KEY, tenant_id INTEGER NOT NULL, sku VARCHAR(64) NOT NULL,
  price_cents INTEGER NOT NULL, quantity INTEGER NOT NULL,
  line_total INTEGER GENERATED ALWAYS AS (price_cents * quantity) STORED,
  is_active BOOLEAN NOT NULL,
  CONSTRAINT uq_product_tenant_sku UNIQUE (tenant_id, sku),
  CONSTRAINT ck_product_nonnegative_price CHECK (price_cents >= 0)
);
CREATE INDEX ix_product_tenant_active ON product (tenant_id, is_active);

from datetime import date, datetime
from decimal import Decimal
from enum import Enum
from uuid import UUID, uuid4
from sqlalchemy import JSON, Date, DateTime, Enum as SAEnum, Numeric, Uuid

class OrderState(str, Enum):
    pending = "pending"
    paid = "paid"

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
    state: Mapped[OrderState] = mapped_column(SAEnum(OrderState))
    amount: Mapped[Decimal] = mapped_column(Numeric(12, 2))
    metadata_: Mapped[dict] = mapped_column("metadata", JSON, default=dict)
    ordered_on: Mapped[date] = mapped_column(Date)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))

CREATE TYPE orderstate AS ENUM ('pending', 'paid');
CREATE TABLE orders (
  id UUID PRIMARY KEY, state orderstate NOT NULL,
  amount NUMERIC(12, 2) NOT NULL, metadata JSON NOT NULL,
  ordered_on DATE NOT NULL, created_at TIMESTAMPTZ NOT NULL
);

from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.types import String, TypeDecorator

settings: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON), default=dict)
profile.settings["theme"] = "dark"  # Session detects in-place edit

class LowercaseEmail(TypeDecorator[str]):
    impl = String(255)
    cache_ok = True
    def process_bind_param(self, value, dialect):
        return value.lower().strip() if value else value

email: Mapped[str] = mapped_column(LowercaseEmail())

UPDATE profile SET settings = '{"theme":"dark"}' WHERE id = :id;
INSERT INTO contact (email) VALUES ('ada@example.com');
-- TypeDecorator converts at Python/database boundary.

from sqlalchemy import MetaData, Table, inspect, select

metadata = MetaData()
legacy_users = Table("legacy_users", metadata, autoload_with=engine)
rows = engine.connect().execute(
    select(legacy_users.c.id, legacy_users.c.email)
)

i = inspect(engine)
print(i.get_table_names())
print(i.get_columns("legacy_users"))
print(i.get_foreign_keys("legacy_users"))
print(i.get_indexes("legacy_users"))

SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = 'legacy_users';

SELECT id, email FROM legacy_users;

from sqlalchemy.ext.automap import automap_base

AutomapBase = automap_base()
AutomapBase.prepare(autoload_with=engine)
LegacyUser = AutomapBase.classes.legacy_users

with Session(engine) as session:
    users = session.scalars(select(LegacyUser).limit(10)).all()

SELECT * FROM legacy_users LIMIT 10;

from sqlalchemy import select
from sqlalchemy.ext.asyncio import (
    async_sessionmaker, create_async_engine,
)

engine = create_async_engine(
    "postgresql+asyncpg://app:secret@localhost/shop",
    pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def find_users() -> list[User]:
    async with AsyncSessionLocal() as session:
        result = await session.scalars(
            select(User).where(User.is_active)
        )
        return result.all()

async def create_user() -> User:
    async with AsyncSessionLocal.begin() as session:
        user = User(name="Ada", email="ada@example.com")
        session.add(user)
    return user

SELECT * FROM user_account WHERE is_active = true;
BEGIN;
INSERT INTO user_account (name, email, is_active)
VALUES ('Ada', 'ada@example.com', true);
COMMIT;

from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import selectinload

class Base(AsyncAttrs, DeclarativeBase):
    pass

# Best: eager load response data
users = (await session.scalars(
    select(User).options(selectinload(User.addresses))
)).all()

# Explicit lazy I/O, if relationship was not preloaded:
addresses = await user.awaitable_attrs.addresses

# For strict APIs, configure relationships with lazy="raise".

SELECT * FROM user_account;
SELECT * FROM address WHERE user_id IN (1, 2, 3);
-- await schedules Python; SQL is normal SQL.

from collections.abc import AsyncGenerator
from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel, ConfigDict, EmailStr
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: EmailStr

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    email: str

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

@app.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(data: UserCreate, session: AsyncSession = Depends(get_session)):
    user = User(name=data.name, email=str(data.email))
    session.add(user)
    try:
        await session.commit()
    except Exception:
        await session.rollback()
        raise
    await session.refresh(user)
    return user

@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    user = await session.get(User, user_id)
    if user is None:
        raise HTTPException(404, "User not found")
    return user

-- POST /users
BEGIN;
INSERT INTO user_account (name, email, is_active)
VALUES (:name, :email, true);
COMMIT;
SELECT id, name, email FROM user_account WHERE id = :new_id;

-- GET /users/{id}
SELECT * FROM user_account WHERE id = :user_id;

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Production: run Alembic before deployment instead.
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

# 1. One session per request.
# 2. Eager-load any relationships response serialization needs.
# 3. Return DTOs/Pydantic models, not an open Session.
# 4. Translate expected IntegrityError to a clear HTTP error.
# 5. Do not run sync database I/O inside async endpoints.