Examples

These are the runnable scripts in examples/, from simple to complex. CI runs every one of them.

01 Basics

"""Basics: filter, project and pick elements from a plain list of pydantic models."""

from pydantic import BaseModel

import pydantic_linq  # noqa: F401  (adds the LINQ operators to list / tuple / set)
from pydantic_linq import Enumerable


class Product(BaseModel):
    name: str
    price: float
    stock: int


products = [
    Product(name="Keyboard", price=49.0, stock=12),
    Product(name="Mouse", price=19.5, stock=0),
    Product(name="Monitor", price=219.0, stock=4),
    Product(name="Cable", price=5.0, stock=120),
]

# Where + Select, called straight on the list
in_stock = products.where(lambda p: p.stock > 0).select(lambda p: p.name).to_list()
print("in stock:", in_stock)

# Element operators
print("first over $100:", products.first(lambda p: p.price > 100).name)
print("first over $1000:", products.first_or_default(lambda p: p.price > 1000))

# Quantifiers and counting. list.count(x) keeps its Python meaning, so wrap for the LINQ count().
print("any sold out:", products.any(lambda p: p.stock == 0))
print("sold out count:", Enumerable(products).count(lambda p: p.stock == 0))

# Queries are deferred: nothing runs until you enumerate
cheap = products.where(lambda p: p.price < 20)
products.append(Product(name="Sticker", price=1.0, stock=500))
print("cheap (evaluated now):", cheap.select(lambda p: p.name).to_list())

02 Sorting and paging

"""Sorting and paging: OrderBy / ThenBy, Skip / Take, Chunk."""

from pydantic import BaseModel

import pydantic_linq  # noqa: F401


class Employee(BaseModel):
    name: str
    dept: str
    salary: int


staff = [
    Employee(name="Ava", dept="Eng", salary=120),
    Employee(name="Ben", dept="Ops", salary=90),
    Employee(name="Cleo", dept="Eng", salary=120),
    Employee(name="Dan", dept="Eng", salary=105),
    Employee(name="Eve", dept="Ops", salary=95),
]

# Multi-key, stable sort: dept ascending, then salary descending, then name
ordered = staff.order_by(lambda e: e.dept).then_by_descending(lambda e: e.salary).then_by(lambda e: e.name)
print([f"{e.dept}:{e.name}:{e.salary}" for e in ordered])

# Paging
page_size = 2
for page in range(3):
    names = ordered.skip(page * page_size).take(page_size).select(lambda e: e.name).to_list()
    print(f"page {page + 1}:", names)

# Chunk does the same in one call
print("chunks:", ordered.select(lambda e: e.name).chunk(2).to_list())

# Take with a range: slice(1, -1) is C#'s Take(1..^1)
print("without top and bottom:", ordered.take(slice(1, -1)).select(lambda e: e.name).to_list())

03 Grouping and aggregation

"""Grouping and aggregation: GroupBy, ToLookup, CountBy, AggregateBy, Sum / Average / MaxBy."""

from datetime import date

from pydantic import BaseModel

import pydantic_linq  # noqa: F401


class Sale(BaseModel):
    region: str
    rep: str
    amount: float
    day: date


sales = [
    Sale(region="North", rep="Ann", amount=120.0, day=date(2026, 1, 3)),
    Sale(region="North", rep="Bob", amount=80.0, day=date(2026, 1, 4)),
    Sale(region="South", rep="Cid", amount=200.0, day=date(2026, 1, 4)),
    Sale(region="North", rep="Ann", amount=60.0, day=date(2026, 2, 1)),
    Sale(region="South", rep="Dee", amount=40.0, day=date(2026, 2, 2)),
]

# GroupBy with a result selector: one summary row per region
summary = sales.group_by(
    lambda s: s.region,
    result_selector=lambda region, items: (region, items.count(), items.sum(lambda s: s.amount)),
).to_list()
print("region, orders, total:", summary)

# Each Grouping is itself an Enumerable with a .key
for g in sales.group_by(lambda s: s.day.month):
    print(f"month {g.key}: avg {g.average(lambda s: s.amount):.2f}, reps {g.select(lambda s: s.rep).distinct().to_list()}")

# A Lookup is an indexable group set; a missing key gives an empty sequence
by_rep = sales.to_lookup(lambda s: s.rep, lambda s: s.amount)
print("Ann:", by_rep["Ann"].to_list(), "Zed:", by_rep["Zed"].to_list())

# .NET 9 CountBy / AggregateBy
print("orders per rep:", sales.count_by(lambda s: s.rep).to_list())
print("revenue per rep:", sales.aggregate_by(lambda s: s.rep, 0.0, lambda acc, s: acc + s.amount).to_list())

# Scalar aggregates
print("total:", sales.sum(lambda s: s.amount))
print("biggest sale:", sales.max_by(lambda s: s.amount).rep)
print("running product:", [1, 2, 3, 4].aggregate(lambda a, b: a * b))

04 Joins and sets

"""Joins and set operations across two model lists."""

from pydantic import BaseModel

import pydantic_linq  # noqa: F401


class Customer(BaseModel):
    id: int
    name: str


class Order(BaseModel):
    customer_id: int
    item: str


customers = [Customer(id=1, name="Ann"), Customer(id=2, name="Bob"), Customer(id=3, name="Cid")]
orders = [
    Order(customer_id=1, item="book"),
    Order(customer_id=1, item="pen"),
    Order(customer_id=3, item="lamp"),
    Order(customer_id=9, item="ghost"),
]

# Inner join
print("join:", customers.join(orders, lambda c: c.id, lambda o: o.customer_id, lambda c, o: (c.name, o.item)).to_list())

# GroupJoin: every customer with their orders (possibly none)
print(
    "group join:",
    customers.group_join(
        orders, lambda c: c.id, lambda o: o.customer_id, lambda c, os: (c.name, os.select(lambda o: o.item).to_list())
    ).to_list(),
)

# .NET 10 LeftJoin / RightJoin: the missing side is None
print(
    "left join:",
    customers.left_join(
        orders, lambda c: c.id, lambda o: o.customer_id, lambda c, o: (c.name, o.item if o else None)
    ).to_list(),
)
print(
    "right join:",
    customers.right_join(
        orders, lambda c: c.id, lambda o: o.customer_id, lambda c, o: (c.name if c else None, o.item)
    ).to_list(),
)

# Set operations use value equality, so equal (non-frozen) models count as duplicates
batch_a = [Customer(id=1, name="Ann"), Customer(id=2, name="Bob")]
batch_b = [Customer(id=2, name="Bob"), Customer(id=4, name="Dee")]
print("union:", batch_a.union(batch_b).select(lambda c: c.name).to_list())
print("intersect:", batch_a.intersect(batch_b).select(lambda c: c.name).to_list())
print("except:", batch_a.except_(batch_b).select(lambda c: c.name).to_list())

# *By variants compare by a key; the second sequence holds keys
print("customers without orders:", customers.except_by(orders.select(lambda o: o.customer_id), lambda c: c.id)
      .select(lambda c: c.name).to_list())

05 Pydantic integration

"""Pydantic integration: validate raw data with cast(), use Enumerable[Model] as a field, round-trip JSON."""

from pydantic import BaseModel, ValidationError

import pydantic_linq  # noqa: F401
from pydantic_linq import Enumerable


class LineItem(BaseModel):
    sku: str
    qty: int
    unit_price: float


class Invoice(BaseModel):
    number: str
    lines: Enumerable[LineItem]  # validated like list[LineItem]; behaves like LINQ

    @property
    def total(self) -> float:
        return self.lines.sum(lambda line: line.qty * line.unit_price)


# Raw rows, e.g. from an API or CSV: strings get coerced, dicts become models
raw = [
    {"sku": "A-1", "qty": "2", "unit_price": "9.99"},
    {"sku": "B-7", "qty": 1, "unit_price": 120},
]
items = raw.cast(LineItem).to_list()
print("cast:", items)

# strict=True disables coercion
try:
    raw.cast(LineItem, strict=True).to_list()
except ValidationError as e:
    print("strict cast failed:", e.error_count(), "errors")

invoice = Invoice(number="INV-1", lines=items)
print("total:", round(invoice.total, 2))
print("big lines:", invoice.lines.where(lambda line: line.unit_price > 50).select(lambda line: line.sku).to_list())

# Serializes as a plain list; validates back into an Enumerable
payload = invoice.model_dump_json()
print("json:", payload)
again = Invoice.model_validate_json(payload)
print("round-trip type:", type(again.lines).__name__, "count:", again.lines.count())

# Value equality: non-frozen models are unhashable, distinct() still works
dupes = [LineItem(sku="A-1", qty=1, unit_price=1.0)] * 3
print("distinct:", dupes.distinct().count())

# of_type filters by class, like OfType<T>
mixed = [items[0], "noise", 42, items[1]]
print("of_type:", mixed.of_type(LineItem).select(lambda line: line.sku).to_list())

06 Async

"""Async: AsyncEnumerable over async sources, with sync or async selectors."""

import asyncio

from pydantic import BaseModel

import pydantic_linq  # noqa: F401
from pydantic_linq import AsyncEnumerable


class User(BaseModel):
    id: int
    name: str


class Profile(BaseModel):
    user_id: int
    followers: int


async def fetch_users():
    """An async generator source, e.g. a paginated API or a DB cursor."""
    for i, name in enumerate(["ann", "bob", "cid", "dee"], start=1):
        await asyncio.sleep(0.01)
        yield User(id=i, name=name)


async def fetch_profile(user: User) -> Profile:
    """An async selector, the System.Linq.Async SelectAwait overload."""
    await asyncio.sleep(0.01)
    return Profile(user_id=user.id, followers=user.id * 150)


async def is_active(user: User) -> bool:
    await asyncio.sleep(0)
    return user.id != 2


async def main() -> None:
    users = AsyncEnumerable(fetch_users())

    # Async predicate + sync projection
    names = await users.where(is_active).select(lambda u: u.name.title()).to_list()
    print("active:", names)

    # Async generator sources are one-shot (as in Python), so re-wrap for a second query
    users = AsyncEnumerable(fetch_users())
    popular = await (
        users.select(fetch_profile)
        .where(lambda p: p.followers > 200)
        .order_by_descending(lambda p: p.followers)
        .select(lambda p: p.user_id)
        .to_list()
    )
    print("popular ids:", popular)

    # Sync lists enter async with to_async_enumerable(); terminal operators are awaited
    local = [User(id=10, name="eve"), User(id=11, name="fay")]
    print("total followers:", await local.to_async_enumerable().select(fetch_profile).sum(lambda p: p.followers))
    print("first:", (await local.to_async_enumerable().first(lambda u: u.id > 10)).name)

    # Grouping yields AsyncGrouping, an AsyncEnumerable with a .key
    async for g in AsyncEnumerable(fetch_users()).group_by(lambda u: u.id % 2 == 0):
        print("even" if g.key else "odd", await g.select(lambda u: u.name).to_list())


asyncio.run(main())

07 End-to-end sales report

"""End to end: raw JSON in, validated models, async enrichment, joins, grouping, a typed report out."""

import asyncio
import json
from datetime import date

from pydantic import BaseModel

from pydantic_linq import Enumerable

RAW_ORDERS = json.loads("""[
  {"id": 1, "customer": "acme",   "sku": "W-1", "qty": 3,  "day": "2026-03-01"},
  {"id": 2, "customer": "globex", "sku": "G-9", "qty": 1,  "day": "2026-03-01"},
  {"id": 3, "customer": "acme",   "sku": "G-9", "qty": 2,  "day": "2026-03-02"},
  {"id": 4, "customer": "initech","sku": "W-1", "qty": 10, "day": "2026-03-03"},
  {"id": 5, "customer": "acme",   "sku": "X-0", "qty": 1,  "day": "2026-03-03"},
  {"id": 2, "customer": "globex", "sku": "G-9", "qty": 1,  "day": "2026-03-01"}
]""")


class Order(BaseModel):
    id: int
    customer: str
    sku: str
    qty: int
    day: date


class Product(BaseModel):
    sku: str
    name: str
    price: float


class CustomerRow(BaseModel):
    customer: str
    orders: int
    revenue: float
    top_product: str


class Report(BaseModel):
    rows: Enumerable[CustomerRow]
    unknown_skus: list[str]
    best_day: date


CATALOG = [Product(sku="W-1", name="Widget", price=4.5), Product(sku="G-9", name="Gadget", price=30.0)]


async def price_lookup(sku: str) -> float | None:
    """Pretend pricing service; async so it shows the async layer."""
    await asyncio.sleep(0.01)
    return {"W-1": 4.0, "G-9": 28.0}.get(sku)  # live price, may differ from the catalog


async def build_report() -> Report:
    # 1. Validate the raw JSON and drop duplicate records (value equality on models)
    orders = Enumerable(RAW_ORDERS).cast(Order).distinct().to_list()

    # 2. SKUs with no catalog entry: except_by against the catalog keys
    unknown = orders.except_by(CATALOG.select(lambda p: p.sku), lambda o: o.sku).select(lambda o: o.sku).to_list()

    # 3. Join to the catalog; live price from the async service, catalog price as fallback
    async def priced(pair: tuple[Order, Product]) -> tuple[Order, Product, float]:
        order, product = pair
        live = await price_lookup(product.sku)
        return order, product, live if live is not None else product.price

    lines = await (
        orders.join(CATALOG, lambda o: o.sku, lambda p: p.sku, lambda o, p: (o, p))
        .to_async_enumerable()
        .select(priced)
        .to_list()
    )

    # 4. Per-customer rows, highest revenue first
    rows = (
        lines.group_by(lambda line: line[0].customer)
        .select(
            lambda g: CustomerRow(
                customer=g.key,
                orders=g.count(),
                revenue=g.sum(lambda line: line[0].qty * line[2]),
                top_product=g.max_by(lambda line: line[0].qty * line[2])[1].name,
            )
        )
        .order_by_descending(lambda r: r.revenue)
        .then_by(lambda r: r.customer)
        .to_list()
    )

    # 5. Day with the most units shipped
    best_day = orders.aggregate_by(lambda o: o.day, 0, lambda n, o: n + o.qty).max_by(lambda kv: kv[1])[0]

    return Report(rows=rows, unknown_skus=unknown, best_day=best_day)


report = asyncio.run(build_report())
print(report.model_dump_json(indent=2))

08 Entity Framework (SQLModel)

"""Entity Framework style queries over SQLModel: LINQ that runs as SQL.

Needs the extra: pip install pydantic-linq[sqlmodel]
"""

from sqlmodel import Field, Relationship, SQLModel, col, create_engine

from pydantic_linq.orm import DbContext


class Blog(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    posts: list["Post"] = Relationship(back_populates="blog")


class Post(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    likes: int
    blog_id: int | None = Field(default=None, foreign_key="blog.id")
    blog: Blog | None = Relationship(back_populates="posts")


with DbContext(create_engine("sqlite://")) as db:
    db.ensure_created()

    # Add / SaveChanges
    python, dotnet = Blog(name="Python"), Blog(name=".NET")
    db.add_range(
        [
            Post(title="Generics in 3.12", likes=120, blog=python),
            Post(title="Pattern matching", likes=80, blog=python),
            Post(title="LINQ deep dive", likes=300, blog=dotnet),
            Post(title="EF Core 10", likes=210, blog=dotnet),
            Post(title="Records", likes=40, blog=dotnet),
        ]
    )
    print("saved:", db.save_changes())

    posts = db.set(Post)  # DbSet<Post>

    # Lambdas build SQL: use & | ~ for and / or / not.
    popular = posts.where(lambda p: (p.likes > 100) & ~col(p.title).contains("3.12")).order_by_descending(lambda p: p.likes)
    print(popular.to_query_string())
    print("popular:", popular.select(lambda p: p.title).to_list())

    # Paging, element operators and aggregates are SQL too.
    print("page 2:", posts.order_by(lambda p: p.id).skip(2).take(2).select(lambda p: p.title).to_list())
    print("top:", posts.max_by(lambda p: p.likes).title, "| total likes:", posts.sum(lambda p: p.likes))

    # GroupBy + Select -> GROUP BY / HAVING
    per_blog = (
        posts.group_by(lambda p: p.blog_id)
        .where(lambda g: g.count() > 1)
        .select(lambda g: (g.key, g.count(), g.average(lambda p: p.likes)))
    )
    print("per blog:", per_blog.to_list())

    # Join -> INNER JOIN; the result selector picks columns.
    titles = posts.join(db.set(Blog), lambda p: p.blog_id, lambda b: b.id, lambda p, b: (b.name, p.title))
    print("joined:", titles.order_by(lambda t: t[1]).take(3).to_list())

    # Include / ThenInclude load relationships eagerly.
    for blog in db.set(Blog).include(lambda b: b.posts).order_by(lambda b: b.name):
        print(blog.name, [p.title for p in blog.posts])

    # A select SQL can't express runs in memory on the fetched rows (EF client projection).
    print(posts.order_by(lambda p: p.likes).select(lambda p: f"{p.title} ({p.likes})").first())

    # Tracked changes, ExecuteUpdate / ExecuteDelete, raw SQL.
    records = posts.single(lambda p: p.title == "Records")
    records.likes += 1
    db.save_changes()
    print("bumped:", posts.where(lambda p: p.likes < 100).execute_update(lambda p: {p.likes: p.likes + 10}))
    print("deleted:", posts.where(lambda p: p.likes < 60).execute_delete())
    db.save_changes()
    print("raw:", db.from_sql(Post, "SELECT * FROM post WHERE likes > :n", n=200).count())

09 Entity Framework, async

"""Async Entity Framework style queries: an AsyncEngine and EF's *Async terminal operators.

Needs the extra and an async driver: pip install pydantic-linq[sqlmodel] aiosqlite
"""

import asyncio

from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import Field, SQLModel

from pydantic_linq.orm import DbContext


class Order(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    customer: str
    total: float


async def main() -> None:
    engine = create_async_engine("sqlite+aiosqlite://")
    async with DbContext(engine) as db:
        await db.ensure_created_async()
        db.add_range(
            Order(customer=c, total=t)
            for c, t in [("ann", 40.0), ("bob", 15.5), ("ann", 99.9), ("cid", 12.0), ("bob", 70.0)]
        )
        await db.save_changes_async()

        orders = db.set(Order)
        print("big orders:", await orders.where(lambda o: o.total > 50).count_async())
        print("first:", (await orders.order_by(lambda o: o.total).first_async()).customer)
        print("revenue:", await orders.sum_async(lambda o: o.total))

        by_customer = orders.group_by(lambda o: o.customer).select(lambda g: (g.key, g.sum(lambda o: o.total)))
        print("by customer:", await by_customer.to_dictionary_async(lambda kv: kv[0], lambda kv: kv[1]))

        # AsAsyncEnumerable: continue with System.Linq.Async operators in memory.
        print("chunks:", await orders.order_by(lambda o: o.id).as_async_enumerable().chunk(2).count())

        async def notify(order: Order) -> None:
            await asyncio.sleep(0)
            print("  notify", order.customer, order.total)

        await orders.where(lambda o: o.customer == "ann").for_each_async(notify)
        print("deleted:", await orders.where(lambda o: o.total < 20).execute_delete_async())
        await db.save_changes_async()
    await engine.dispose()


asyncio.run(main())