動的なフィルタリングとソートパラメーター

Why Dynamic Query Parameters?

Collecting Filters with a Dependency Class

from typing import Optional
from fastapi import Query

class UserFilterParams:
    def __init__(
        self,
        status: Optional[str] = Query(None, description="active | inactive"),
        min_age: Optional[int] = Query(None, ge=0, le=150),
        search: Optional[str] = Query(None, min_length=2, max_length=50),
    ):
        self.status = status
        self.min_age = min_age
        self.search = search

# Usage:
# @app.get('/users')
# def list_users(filters: UserFilterParams = Depends()):
#     ...

Validating Filter Values with Enums

from enum import Enum
from typing import Optional
from fastapi import Query

class UserStatus(str, Enum):
    active = "active"
    inactive = "inactive"
    pending = "pending"

class UserFilterParams:
    def __init__(
        self,
        status: Optional[UserStatus] = Query(None),
        min_age: Optional[int] = Query(None, ge=0, le=150),
    ):
        self.status = status
        self.min_age = min_age

Turning Filters into Predicates

USERS = [
    {"id": 1, "name": "Ada", "status": "active", "age": 36},
    {"id": 2, "name": "Linus", "status": "inactive", "age": 54},
    {"id": 3, "name": "Grace", "status": "active", "age": 41},
]

def apply_filters(rows, status=None, min_age=None, search=None):
    result = rows
    if status is not None:
        result = [r for r in result if r["status"] == status]
    if min_age is not None:
        result = [r for r in result if r["age"] >= min_age]
    if search is not None:
        result = [r for r in result if search.lower() in r["name"].lower()]
    return result

print(apply_filters(USERS, status="active", min_age=40))

Parsing a Sort Parameter

ALLOWED_SORT = {"created_at", "name", "age", "id"}

def parse_sort(sort_param):
    parsed = []
    for token in sort_param.split(","):
        token = token.strip()
        if not token:
            continue
        descending = token.startswith("-")
        field = token[1:] if descending else token
        if field not in ALLOWED_SORT:
            raise ValueError(f"Cannot sort by '{field}'")
        parsed.append((field, "desc" if descending else "asc"))
    return parsed

print(parse_sort("-created_at,name"))
print(parse_sort("age"))

A Reusable Sort Dependency

from typing import Optional
from fastapi import Query, HTTPException

def sort_dependency(allowed: set):
    def _parse(sort: Optional[str] = Query(None, example="-created_at,name")):
        if not sort:
            return []
        parsed = []
        for token in sort.split(","):
            token = token.strip()
            if not token:
                continue
            desc = token.startswith("-")
            field = token[1:] if desc else token
            if field not in allowed:
                raise HTTPException(422, f"Invalid sort field: {field}")
            parsed.append((field, "desc" if desc else "asc"))
        return parsed
    return _parse

# @app.get('/users')
# def list_users(sort=Depends(sort_dependency({'created_at','name'}))):
#     ...

Applying Multi-Key Sort In Memory

ROWS = [
    {"name": "Ada", "age": 36},
    {"name": "Grace", "age": 36},
    {"name": "Linus", "age": 54},
]

def apply_sort(rows, sort_keys):
    result = list(rows)
    for field, direction in reversed(sort_keys):
        result.sort(key=lambda r: r[field], reverse=(direction == "desc"))
    return result

ordered = apply_sort(ROWS, [("age", "desc"), ("name", "asc")])
for r in ordered:
    print(r)

Field Selection (Sparse Fieldsets)

EXPOSABLE = {"id", "name", "status", "age"}

def select_fields(rows, fields_param):
    if not fields_param:
        return rows
    requested = {f.strip() for f in fields_param.split(",") if f.strip()}
    invalid = requested - EXPOSABLE
    if invalid:
        raise ValueError(f"Unknown fields: {sorted(invalid)}")
    return [{k: r[k] for k in requested if k in r} for r in rows]

data = [{"id": 1, "name": "Ada", "status": "active", "age": 36}]
print(select_fields(data, "id,name"))

Combining Filter, Sort, Select and Pagination

def list_resource(rows, *, filters, sort_keys, fields, offset, limit,
                  apply_filters, apply_sort, select_fields):
    rows = apply_filters(rows, **filters)
    rows = apply_sort(rows, sort_keys)
    total = len(rows)
    page = rows[offset: offset + limit]
    page = select_fields(page, fields)
    return {"total": total, "items": page,
            "offset": offset, "limit": limit}

# In FastAPI each piece is a Depends(); the route just calls list_resource.

Composing Dependencies into One Query Object

from dataclasses import dataclass
from typing import Optional
from fastapi import Depends, Query

@dataclass
class ListQuery:
    filters: object
    sort: list
    fields: Optional[str]
    offset: int
    limit: int

def list_query(
    filters: "UserFilterParams" = Depends(),
    sort: list = Depends(sort_dependency({"created_at", "name"})),
    fields: Optional[str] = Query(None),
    offset: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
) -> ListQuery:
    return ListQuery(filters, sort, fields, offset, limit)

# @app.get('/users')
# def list_users(q: ListQuery = Depends(list_query)):
#     ...

Documenting and Defaulting the Contract

Quick Check: Pipeline Order

Recap