### Install Dependencies (Bash)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

This command installs the necessary Python packages for the project: `uvicorn` for running the ASGI application, `aiosqlite` for asynchronous SQLite support, and `fastsqla` for the API framework. These dependencies are crucial for setting up and running the example application.

```bash
pip install uvicorn aiosqlite fastsqla
```

--------------------------------

### Install FastSQLA

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-setup/SKILL.md

Installs the FastSQLA library using pip. This is the basic installation required for using FastSQLA with FastAPI.

```bash
pip install FastSQLA
```

--------------------------------

### Install FastSQLA with SQLModel Support

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-setup/SKILL.md

Installs FastSQLA with optional support for SQLModel. This enables FastSQLA to automatically use SQLModel's async session class.

```bash
pip install FastSQLA[sqlmodel]
```

--------------------------------

### Run FastAPI Application (Bash)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

This command starts the FastAPI application using `uvicorn`. It sets the `sqlalchemy_url` environment variable to point to the SQLite database and specifies the application module (`example:app`). This command is used to serve the API locally.

```bash
sqlalchemy_url=sqlite+aiosqlite:///db.sqlite?check_same_thread=false \
  uvicorn example:app
```

--------------------------------

### FastSQLAlchemy Quick Example: Full FastAPI Application

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

A complete, runnable example of a FastAPI application using FastSQLAlchemy. It defines a `Hero` model using SQLAlchemy's ORM, sets up the FastAPI app with the lifespan, and includes endpoints for listing, retrieving, and creating heroes with pagination and session management.

```python
# example.py
from http import HTTPStatus

from fastapi import FastAPI, HTTPException
from fastsqla import Base, Item, Page, Paginate, Session, lifespan
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Mapped, mapped_column


app = FastAPI(lifespan=lifespan)


class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    secret_identity: Mapped[str]
    age: Mapped[int]


class HeroBase(BaseModel):
    name: str
    secret_identity: str
    age: int


class HeroModel(HeroBase):
    model_config = ConfigDict(from_attributes=True)
    id: int


@app.get("/heros", response_model=Page[HeroModel])
async def list_heros(paginate: Paginate):
    stmt = select(Hero)
    return await paginate(stmt)


@app.get("/heros/{hero_id}", response_model=Item[HeroModel])
async def get_hero(hero_id: int, session: Session):
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(HTTPStatus.NOT_FOUND, "Hero not found")
    return {"data": hero}


@app.post("/heros", response_model=Item[HeroModel])
async def create_hero(new_hero: HeroBase, session: Session):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    try:
        await session.flush()
    except IntegrityError:
        raise HTTPException(HTTPStatus.CONFLICT, "Duplicate hero name")
    return {"data": hero}
```

--------------------------------

### Compose Multiple Lifespans with AsyncExitStack (Python)

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-setup/SKILL.md

Demonstrates how to compose multiple lifespan contexts, such as FastSQLA's lifespan with another library's lifespan, using `AsyncExitStack`. This ensures all contexts are properly managed during startup and shutdown.

```python
from collections.abc import AsyncGenerator
from contextlib import AsyncExitStack, asynccontextmanager

from fastapi import FastAPI
from fastsqla import lifespan as fastsqla_lifespan
from other_library import lifespan as other_lifespan


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[dict, None]:
    async with AsyncExitStack() as stack:
        state1 = await stack.enter_async_context(fastsqla_lifespan(app))
        state2 = await stack.enter_async_context(other_lifespan(app))
        yield {**state1, **state2}


app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Install FastSQLAlchemy using uv

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Command to install the FastSQLAlchemy package using the `uv` package installer. This is a recommended method for managing Python dependencies.

```bash
uv add fastsqla
```

--------------------------------

### Configure FastSQLA using Environment Variables (Bash)

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-setup/SKILL.md

Sets environment variables for FastSQLA configuration. These variables are read by `fastsqla.lifespan` to configure the SQLAlchemy engine.

```bash
export SQLALCHEMY_URL=postgresql+asyncpg://user:pass@localhost/mydb
export SQLALCHEMY_POOL_SIZE=20
export SQLALCHEMY_MAX_OVERFLOW=10
export SQLALCHEMY_ECHO=true
```

--------------------------------

### Complete FastSQLA Application Example

Source: https://context7.com/hadrien/fastsqla/llms.txt

A comprehensive example demonstrating FastSQLA features including CRUD operations, pagination, error handling, and model definitions using SQLAlchemy and Pydantic. This example provides a robust foundation for API development. It requires `fastapi`, `fastsqla`, `pydantic`, and `sqlalchemy`.

```python
from http import HTTPStatus
from fastapi import FastAPI, HTTPException
from fastsqla import Base, Item, Page, Paginate, Session, lifespan
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI(lifespan=lifespan)

# Model definition
class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    secret_identity: Mapped[str]
    age: Mapped[int]

# Pydantic schemas
class HeroBase(BaseModel):
    name: str
    secret_identity: str
    age: int

class HeroModel(HeroBase):
    model_config = ConfigDict(from_attributes=True)
    id: int

# List with pagination
@app.get("/heroes", response_model=Page[HeroModel])
async def list_heroes(paginate: Paginate):
    return await paginate(select(Hero))

# Get single item
@app.get("/heroes/{hero_id}", response_model=Item[HeroModel])
async def get_hero(hero_id: int, session: Session):
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(HTTPStatus.NOT_FOUND, "Hero not found")
    return {"data": hero}

# Create with duplicate handling
@app.post("/heroes", response_model=Item[HeroModel])
async def create_hero(new_hero: HeroBase, session: Session):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    try:
        await session.flush()
    except IntegrityError:
        raise HTTPException(HTTPStatus.CONFLICT, "Hero name already exists")
    return {"data": hero}

# Update
@app.put("/heroes/{hero_id}", response_model=Item[HeroModel])
async def update_hero(hero_id: int, hero_data: HeroBase, session: Session):
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(HTTPStatus.NOT_FOUND, "Hero not found")
    for key, value in hero_data.model_dump().items():
        setattr(hero, key, value)
    await session.flush()
    return {"data": hero}

# Delete
@app.delete("/heroes/{hero_id}")
async def delete_hero(hero_id: int, session: Session):
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(HTTPStatus.NOT_FOUND, "Hero not found")
    await session.delete(hero)
    return {"message": "Hero deleted"}
```

--------------------------------

### Install FastSQLAlchemy using pip

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Command to install the FastSQLAlchemy package using `pip`, the standard Python package installer. This is a common method for adding libraries to your project.

```bash
pip install fastsqla
```

--------------------------------

### Configure PostgreSQL Connection via Environment Variables (Bash)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/setup.md

Sets the PostgreSQL database URL using the asyncpg driver and configures a pool recycle time of 30 minutes. This demonstrates environment variable configuration for a specific database dialect.

```bash
export SQLALCHEMY_URL=postgresql+asyncpg://postgres@localhost/postgres
export SQLALCHEMY_POOL_RECYCLE=1800
```

--------------------------------

### Programmatic FastSQLA Configuration (Python)

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-setup/SKILL.md

Configures FastSQLA programmatically using `new_lifespan()`. This method allows passing configuration arguments directly in code, similar to SQLAlchemy's `create_async_engine()`.

```python
from fastapi import FastAPI
from fastsqla import new_lifespan

lifespan = new_lifespan(
    "sqlite+aiosqlite:///app/db.sqlite",
    connect_args={"autocommit": False},
)

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Configure FastSQLA using Environment Variables (Python)

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-setup/SKILL.md

Configures FastSQLA using environment variables for a 12-factor app style. The `fastsqla.lifespan` context manager reads configuration from environment variables prefixed with SQLALCHEMY_.

```python
from fastapi import FastAPI
from fastsqla import lifespan

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Custom Page Sizes Example

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Illustrates how to create a pagination dependency with custom minimum and maximum page sizes using `new_pagination()`.

```APIDOC
## Custom Page Sizes

### Endpoint
`GET /heroes`

### Code Example
```python
from typing import Annotated
from fastapi import Depends, FastAPI
from fastsqla import Base, Page, PaginateType, new_pagination
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI()

# ... (Hero and HeroModel definitions as above) ...

SmallPagePaginate = Annotated[
    PaginateType[HeroModel],
    Depends(new_pagination(min_page_size=5, max_page_size=25)),
]

@app.get("/heroes")
async def list_heroes(paginate: SmallPagePaginate) -> Page[HeroModel]:
    return await paginate(select(Hero))
```

This endpoint has `limit` defaulting to `5` with a maximum of `25`.
```

--------------------------------

### Setup FastAPI App with Lifespan

Source: https://context7.com/hadrien/fastsqla/llms.txt

Demonstrates how to initialize a FastAPI application using the `lifespan` context manager from FastSQLA. This method automatically configures the database engine and connection pool based on the SQLALCHEMY_URL environment variable.

```python
from fastapi import FastAPI
from fastsqla import lifespan

# Basic setup - reads SQLALCHEMY_URL from environment
app = FastAPI(lifespan=lifespan)

# Run with environment variable:
# SQLALCHEMY_URL=postgresql+asyncpg://user:pass@localhost/db uvicorn app:app
```

--------------------------------

### SQLModel Pagination Example

Source: https://github.com/hadrien/fastsqla/blob/main/docs/pagination.md

Shows how to paginate SQLModel query results within a FastAPI application. This example leverages `fastapi.Page` and `fastsqla.Paginate` for simplified pagination. The endpoint is defined to return a `Page` of `Hero` models, and the `Paginate` dependency injects an async function to perform the pagination when awaited with a SQLAlchemy select statement.

```python
from fastapi import FastAPI
from fastsqla import Page, Paginate, Session
from sqlmodel import Field, SQLModel
from sqlalchemy import select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_identity: str
    age: int


@app.get("/heroes", response_model=Page[Hero])
async def get_heroes(paginate: Paginate):
    return await paginate(select(Hero))

```

--------------------------------

### Configure MariaDB Connection via Environment Variables (Bash)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/setup.md

Sets the MariaDB database URL using the aiomysql driver and enables echo logging for SQLAlchemy. This illustrates environment variable configuration for MariaDB with specific driver parameters.

```bash
export sqlalchemy_url=mysql+aiomysql://bob:password!@db.example.com/app
export sqlalchemy_echo=true
```

--------------------------------

### SQLModel Integration with FastSQLAlchemy

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Demonstrates the seamless integration of SQLModel with FastSQLAlchemy for defining database models and API endpoints. Includes examples for fetching paginated lists and individual items.

```python
from fastsqla import Item, Page, Paginate, Session
from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_identity: str
    age: int

@app.get("/heroes", response_model=Page[Hero])
async def get_heroes(paginate: Paginate):
    return await paginate(select(Hero))

@app.get("/heroes/{hero_id}", response_model=Item[Hero])
async def get_hero(session: Session, hero_id: int):
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
    return {"data": hero}
```

--------------------------------

### Configure SQLite Connection via Environment Variables (Bash)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/setup.md

Sets the SQLite database file path using the aiosqlite driver and configures a connection pool size of 50. This showcases environment variable configuration for a local SQLite database.

```bash
export sqlalchemy_url=sqlite+aiosqlite:///tmp/test.db
export sqlalchemy_pool_size=50
```

--------------------------------

### FastAPI App Startup with FastSQLAlchemy Lifespan

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Integrates FastSQLAlchemy's lifespan events into a FastAPI application for easy setup at application startup. This ensures proper initialization and cleanup of database connections.

```python
from fastapi import FastAPI
from fastsqla import lifespan

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Full Custom Pagination with Joins and Custom Processors

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Provides a complete example of custom pagination combining a custom count query and a custom result processor for handling joins. This setup is useful for complex queries returning multiple related entities.

```python
from typing import Annotated, cast
from fastapi import Depends, FastAPI
from fastsqla import Base, Page, PaginateType, Session, new_pagination
from pydantic import BaseModel
from sqlalchemy import ForeignKey, String, func, select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI()

class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True)
    name: Mapped[str]

class Sticky(Base):
    __tablename__ = "sticky"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey(User.id))
    body: Mapped[str]

class StickyModel(BaseModel):
    id: int
    body: str
    user_id: int
    user_email: str
    user_name: str

async def query_count(session: Session) -> int:
    result = await session.execute(select(func.count()).select_from(Sticky))
    return cast(int, result.scalar())

CustomPaginate = Annotated[
    PaginateType[StickyModel],
    Depends(
        new_pagination(
            query_count_dependency=query_count,
            result_processor=lambda result: iter(result.mappings()),
        )
    ),
]

@app.get("/stickies")
async def list_stickies(paginate: CustomPaginate) -> Page[StickyModel]:
    stmt = select(
        Sticky.id,
        Sticky.body,
        User.id.label("user_id"),
        User.email.label("user_email"),
        User.name.label("user_name"),
    ).join(User)
    return await paginate(stmt)
```

--------------------------------

### Fetch Heroes Data with Pagination (Curl)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

This `curl` command sends a GET request to the `/heros` endpoint of the running FastAPI application. It includes `offset` and `limit` query parameters to demonstrate pagination, fetching a specific subset of hero data. The `accept` header is set to `application/json` to request a JSON response.

```bash
curl -X 'GET' -H 'accept: application/json' 'http://127.0.0.1:8000/heros?offset=10&limit=10'
```

--------------------------------

### Custom Count Query Example

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Demonstrates how to provide a custom dependency to `new_pagination()` for calculating the total item count, useful for complex queries.

```APIDOC
## Custom Count Query

The default count query runs `SELECT COUNT(*) FROM (your_select_as_subquery)`. For joins or complex queries where this is inefficient, provide a `query_count_dependency` — a FastAPI dependency that receives the session and returns an `int`:

### Code Example for `query_count`
```python
from typing import cast
from sqlalchemy import func, select
from fastsqla import Session

# Assuming 'Sticky' and 'StickyModel' are defined elsewhere
# class Sticky(Base):
#     __tablename__ = "sticky"
#     id: Mapped[int] = mapped_column(primary_key=True)
#     # ... other fields

async def query_count(session: Session) -> int:
    result = await session.execute(select(func.count()).select_from(Sticky))
    return cast(int, result.scalar())
```

### Code Example for Custom Dependency
```python
from typing import Annotated
from fastapi import Depends, FastAPI
from fastsqla import Base, Page, PaginateType, new_pagination
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI()

# ... (Sticky and StickyModel definitions) ...

# ... (query_count function definition) ...

CustomPaginate = Annotated[
    PaginateType[StickyModel],
    Depends(new_pagination(query_count_dependency=query_count)),
]

@app.get("/stickies")
async def list_stickies(paginate: CustomPaginate) -> Page[StickyModel]:
    stmt = select(Sticky)
    return await paginate(stmt)
```
```

--------------------------------

### GET /heros

Source: https://github.com/hadrien/fastsqla/blob/main/README.md

Retrieves a list of heroes from the database with optional pagination.

```APIDOC
## GET /heros

### Description
Retrieves a list of heroes from the database. Supports pagination using `offset` and `limit` query parameters.

### Method
GET

### Endpoint
/heros

### Parameters
#### Query Parameters
- **offset** (integer) - Optional - The number of items to skip before starting to collect the result set.
- **limit** (integer) - Optional - The numbers of items to return.

### Request Example
```bash
curl -X 'GET' -H 'accept: application/json' 'http://127.0.0.1:8000/heros?offset=10&limit=10'
```

### Response
#### Success Response (200)
- **data** (array) - An array of hero objects.
  - **name** (string) - The name of the hero.
  - **secret_identity** (string) - The secret identity of the hero.
  - **id** (integer) - The unique identifier of the hero.
- **meta** (object) - Metadata about the pagination.
  - **offset** (integer) - The offset used for the request.
  - **total_items** (integer) - The total number of items available.
  - **total_pages** (integer) - The total number of pages.
  - **page_number** (integer) - The current page number.

#### Response Example
```json
{
  "data": [
    {
      "name": "The Flash",
      "secret_identity": "Barry Allen",
      "id": 11
    },
    {
      "name": "Green Lantern",
      "secret_identity": "Hal Jordan",
      "id": 12
    }
  ],
  "meta": {
    "offset": 10,
    "total_items": 12,
    "total_pages": 2,
    "page_number": 2
  }
}
```
```

--------------------------------

### SQLModel Integration for Combined ORM and Response Models

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Demonstrates how to use SQLModel with fastsqla, where models serve as both ORM and response models. This simplifies the setup by eliminating the need for separate Pydantic models.

```python
from fastsqla import Page, Paginate
from sqlmodel import Field, SQLModel, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(unique=True)
    age: int

@app.get("/heroes")
async def list_heroes(paginate: Paginate[Hero]) -> Page[Hero]:
    return await paginate(select(Hero))
```

--------------------------------

### GET /heroes - List Heroes with Pagination (SQLModel)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/pagination.md

Retrieves a paginated list of heroes using SQLModel.

```APIDOC
## GET /heroes

### Description
Retrieves a paginated list of heroes using SQLModel. The response is a `Page` object containing `Hero` instances.

### Method
GET

### Endpoint
/heroes

### Parameters
#### Query Parameters
- **paginate** (Paginate) - Required - Injected by FastSQLA to handle pagination logic.

### Request Example
```
GET /heroes?page=1&size=10
```

### Response
#### Success Response (200)
- **Page[Hero]** - A Page object containing a list of Hero objects and pagination metadata.

#### Response Example
```json
{
  "items": [
    {
      "id": 1,
      "name": "Deadpond",
      "secret_identity": "Dive Wilson",
      "age": 30
    }
  ],
  "total": 100,
  "page": 1,
  "size": 10,
  "pages": 10
}
```
```

--------------------------------

### Create Hero with Correct flush() Usage

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-session/SKILL.md

Shows the correct way to create a new hero using `session.flush()` to get server-generated data (like `hero.id`) before the endpoint returns. FastSQLA will automatically commit the transaction upon successful completion.

```python
from fastsqla import Session, Item

@app.post("/heroes", response_model=Item[HeroItem])
async def create_hero(session: Session, new_hero: HeroBase):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    await session.flush()  # hero.id is now populated
    return {"data": hero}
# FastSQLA auto-commits here
```

--------------------------------

### Example API Response for Heroes Endpoint

Source: https://github.com/hadrien/fastsqla/blob/main/README.md

This JSON object represents the expected response from the `/heros` API endpoint when querying with `offset=10` and `limit=10`. It includes a 'data' array containing hero objects and a 'meta' object with pagination details.

```json
{
  "data": [
    {
      "name": "The Flash",
      "secret_identity": "Barry Allen",
      "id": 11
    },
    {
      "name": "Green Lantern",
      "secret_identity": "Hal Jordan",
      "id": 12
    }
  ],
  "meta": {
    "offset": 10,
    "total_items": 12,
    "total_pages": 2,
    "page_number": 2
  }
}
```

--------------------------------

### GET /heros - List Heroes with Pagination

Source: https://github.com/hadrien/fastsqla/blob/main/docs/pagination.md

Retrieves a paginated list of heroes. Supports filtering by age.

```APIDOC
## GET /heros

### Description
Retrieves a paginated list of heroes. The response is a `Page` object containing `HeroModel` instances. This endpoint supports optional filtering by age.

### Method
GET

### Endpoint
/heros

### Parameters
#### Query Parameters
- **paginate** (Paginate) - Required - Injected by FastSQLA to handle pagination logic.
- **age** (integer) - Optional - Filters the heroes by age.

### Request Example
```
GET /heros?page=1&size=10&age=30
```

### Response
#### Success Response (200)
- **Page[HeroModel]** - A Page object containing a list of HeroModel objects and pagination metadata.

#### Response Example
```json
{
  "items": [
    {
      "id": 1,
      "name": "Deadpond",
      "secret_identity": "Dive Wilson",
      "age": 30
    }
  ],
  "total": 100,
  "page": 1,
  "size": 10,
  "pages": 10
}
```
```

--------------------------------

### SQLAlchemy Pagination Example

Source: https://github.com/hadrien/fastsqla/blob/main/docs/pagination.md

Demonstrates how to paginate SQLAlchemy query results in a FastAPI endpoint. It uses `fastapi.Page` and `fastsqla.Paginate` to automatically handle pagination logic. The endpoint returns a `Page` model of `HeroModel` and accepts a `Paginate` argument which provides an async pagination function. Filtering can be added by including query parameters in the endpoint definition.

```python
from fastapi import FastAPI
from fastsqla import Base, Paginate, Page, lifespan
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI(lifespan=lifespan)

class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    secret_identity: Mapped[str]
    age: Mapped[int]


class HeroModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    secret_identity: str
    age: int


@app.get("/heros", response_model=Page[HeroModel]) # (1)!
async def list_heros(paginate: Paginate): # (2)!
    return await paginate(select(Hero)) # (3)!

```

```python
@app.get("/heros", response_model=Page[HeroModel])
async def list_heros(paginate: Paginate, age:int | None = None):
    stmt = select(Hero)
    if age:
        stmt = stmt.where(Hero.age == age)
    return await paginate(stmt)

```

--------------------------------

### Async Session Dependency Injection in FastAPI Endpoints

Source: https://context7.com/hadrien/fastsqla/llms.txt

Demonstrates using the `Session` dependency in FastAPI endpoints to automatically manage SQLAlchemy sessions. The session is committed on success, rolled back on failure, and closed after each request. Includes examples for retrieving, creating, and querying data.

```python
from http import HTTPStatus
from fastapi import FastAPI, HTTPException
from fastsqla import Session, Item, lifespan
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select

app = FastAPI(lifespan=lifespan)

class HeroModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    secret_identity: str
    age: int

class HeroCreate(BaseModel):
    name: str
    secret_identity: str
    age: int

@app.get("/heroes/{hero_id}", response_model=Item[HeroModel])
async def get_hero(session: Session, hero_id: int):
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(HTTPStatus.NOT_FOUND, "Hero not found")
    return {"data": hero}

@app.post("/heroes", response_model=Item[HeroModel])
async def create_hero(session: Session, new_hero: HeroCreate):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    await session.flush()  # Get auto-generated ID without committing
    return {"data": hero}

# Query example
@app.get("/heroes/search/{name}")
async def search_heroes(session: Session, name: str):
    stmt = select(Hero).where(Hero.name.ilike(f"%{name}%"))
    result = await session.execute(stmt)
    return {"data": list(result.scalars())}
```

--------------------------------

### Non-Paginated Collection with FastSQLA

Source: https://context7.com/hadrien/fastsqla/llms.txt

Demonstrates how to use the `Collection` model for non-paginated lists. This is useful for endpoints that return all matching items without pagination metadata. It requires `fastsqla`, `sqlalchemy`, and a FastAPI application setup.

```python
from fastsqla import Collection, Session
from sqlalchemy import select

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

@app.get("/tags", response_model=Collection[TagModel])
async def list_all_tags(session: Session):
    result = await session.execute(select(Tag))
    tags = result.scalars().all()
    return {"data": tags}

# Response: {"data": [{"id": 1, "name": "python"}, {"id": 2, "name": "fastapi"}]}
```

--------------------------------

### Create SQLite Database and Insert Data (Bash)

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

This snippet uses the `sqlite3` command-line tool to create an SQLite database file named `db.sqlite`. It defines a `hero` table with columns for id, name, secret_identity, and age, and then inserts 12 rows of sample hero data into the table. This is a foundational step for demonstrating database interactions.

```bash
sqlite3 db.sqlite <<EOF
-- Create Table hero
CREATE TABLE hero (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    name            TEXT NOT NULL UNIQUE, -- Unique hero name (e.g., Superman)
    secret_identity TEXT NOT NULL,        -- Secret identity (e.g., Clark Kent)
    age             INTEGER NOT NULL      -- Age of the hero (e.g., 30)
);

-- Insert heroes with their name, secret identity, and age
INSERT INTO hero (name, secret_identity, age) VALUES ('Superman',        'Clark Kent',       30);
INSERT INTO hero (name, secret_identity, age) VALUES ('Batman',          'Bruce Wayne',      35);
INSERT INTO hero (name, secret_identity, age) VALUES ('Wonder Woman',    'Diana Prince',     30);
INSERT INTO hero (name, secret_identity, age) VALUES ('Iron Man',        'Tony Stark',       45);
INSERT INTO hero (name, secret_identity, age) VALUES ('Spider-Man',      'Peter Parker',     25);
INSERT INTO hero (name, secret_identity, age) VALUES ('Captain America', 'Steve Rogers',     100);
INSERT INTO hero (name, secret_identity, age) VALUES ('Black Widow',     'Natasha Romanoff', 35);
INSERT INTO hero (name, secret_identity, age) VALUES ('Thor',            'Thor Odinson',     1500);
INSERT INTO hero (name, secret_identity, age) VALUES ('Scarlet Witch',   'Wanda Maximoff',   30);
INSERT INTO hero (name, secret_identity, age) VALUES ('Doctor Strange',  'Stephen Strange',  40);
INSERT INTO hero (name, secret_identity, age) VALUES ('The Flash',       'Barry Allen',      28);
INSERT INTO hero (name, secret_identity, age) VALUES ('Green Lantern',   'Hal Jordan',       35);
EOF
```

--------------------------------

### Issue Tracking with bd

Source: https://github.com/hadrien/fastsqla/blob/main/AGENTS.md

Provides quick reference commands for the 'bd' (beads) tool used for issue tracking in the project. These commands allow agents to find, view, update, and close issues, as well as sync with Git.

```bash
bd ready              # Find available work
bd show <id>          # View issue details
bd update <id> --status in_progress  # Claim work
bd close <id>         # Complete work
bd sync               # Sync with git
```

--------------------------------

### SQLAlchemy Async Session with Context Manager

Source: https://github.com/hadrien/fastsqla/blob/main/README.md

Shows how to obtain and use an async SQLAlchemy session within a context manager using `fastsqla.open_session`. This is useful for managing sessions outside of FastAPI request contexts, such as in background tasks.

```python
from fastsqla import open_session

async def background_job():
    async with open_session() as session:
        stmt = select(...)
        result = await session.execute(stmt)
        ...
```

--------------------------------

### Customizing Pagination with `new_pagination()`

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Explains how to use the `new_pagination()` factory to create custom pagination dependencies with configurable page sizes and result processing.

```APIDOC
## The `new_pagination()` Factory

For custom pagination behavior, use `new_pagination()` to create a new dependency. It accepts four parameters:

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `min_page_size` | `int` | `10` | Default and minimum `limit` value |
| `max_page_size` | `int` | `100` | Maximum allowed `limit` value |
| `query_count_dependency` | `Callable[..., Awaitable[int]] | None` | `None` | FastAPI dependency returning total item count. When `None`, uses `SELECT COUNT(*) FROM (subquery)`. |
| `result_processor` | `Callable[[Result], Iterable]` | `lambda r: iter(r.unique().scalars())` | Transforms the SQLAlchemy `Result` into an iterable of items |

The return value is a FastAPI dependency. Use it with `Annotated` and `Depends`.
```

--------------------------------

### SQLAlchemy Async Session with Async Context Manager

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Shows how to manually manage an asynchronous SQLAlchemy session using FastSQLAlchemy's `open_session` context manager. This is useful for operations outside of typical request/response cycles, like background tasks.

```python
from fastsqla import open_session
from sqlalchemy import select

async def background_job():
    async with open_session() as session:
        stmt = select(...)
        result = await session.execute(stmt)
        ...
```

--------------------------------

### Basic Usage with Paginate Dependency

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Demonstrates how to use the built-in `Paginate` dependency to add offset and limit query parameters to a FastAPI endpoint.

```APIDOC
## Basic Usage with `Paginate`

`Paginate` is a pre-configured FastAPI dependency. It injects a callable that accepts a SQLAlchemy `Select` and returns a `Page`.

Default query parameters added to the endpoint:
- `offset`: int, default `0`, minimum `0`
- `limit`: int, default `10`, minimum `1`, maximum `100`

### Endpoint
`GET /heroes`

### Request Example
`GET /heroes?offset=20&limit=10`

### Code Example
```python
from fastapi import FastAPI
from fastsqla import Base, Page, Paginate
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI()

class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    age: Mapped[int]

class HeroModel(BaseModel):
    id: int
    name: str
    age: int

@app.get("/heroes")
async def list_heroes(paginate: Paginate[HeroModel]) -> Page[HeroModel]:
    return await paginate(select(Hero))
```

### Response Example (Success 200)
```json
{
  "data": [
    { "id": 1, "name": "Deadpond", "age": 30 },
    { "id": 2, "name": "Spider-Boy", "age": 16 }
  ],
  "meta": {
    "offset": 0,
    "total_items": 42,
    "total_pages": 5,
    "page_number": 1
  }
}
```
```

--------------------------------

### Custom Pagination Dependency with New Factory

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Illustrates creating a custom pagination dependency using the `new_pagination()` factory. This allows for customization of `min_page_size` and `max_page_size`, providing more control over pagination behavior.

```python
from typing import Annotated
from fastapi import Depends
from fastsqla import Page, PaginateType, new_pagination
from pydantic import BaseModel
from sqlalchemy import select
from fastsqla import Base
from sqlalchemy.orm import Mapped

class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int]
    name: Mapped[str]
    age: Mapped[int]

class HeroModel(BaseModel):
    id: int
    name: str
    age: int

SmallPagePaginate = Annotated[
    PaginateType[HeroModel],
    Depends(new_pagination(min_page_size=5, max_page_size=25)),
]

# Assuming 'app' is your FastAPI instance
# @app.get("/heroes")
# async def list_heroes(paginate: SmallPagePaginate) -> Page[HeroModel]:
#     return await paginate(select(Hero))

```

--------------------------------

### Get User Endpoint with Session Dependency

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-session/SKILL.md

Demonstrates how to use the `Session` dependency in a FastAPI endpoint to retrieve a user. The `Session` object is automatically injected, and FastSQLA manages its lifecycle (commit/rollback/close).

```python
from fastsqla import Session, Item

@app.get("/users/{user_id}", response_model=Item[UserModel])
async def get_user(session: Session, user_id: int):
    user = await session.get(User, user_id)
    return {"data": user}
```

--------------------------------

### Mandatory Git Workflow for Session Completion

Source: https://github.com/hadrien/fastsqla/blob/main/AGENTS.md

Details the critical steps required to complete a work session, ensuring all work is tracked, quality gates are passed, issues are updated, and code is successfully pushed to the remote repository. This workflow emphasizes Git commands for pulling, syncing, and pushing.

```bash
git pull --rebase
bd sync
git push
git status  # MUST show "up to date with origin"
```

--------------------------------

### FastSQLAlchemy Pagination Customization

Source: https://github.com/hadrien/fastsqla/blob/main/README.md

Allows customization of pagination parameters like minimum and maximum page sizes using `new_pagination`. This provides flexibility in controlling the pagination behavior of API endpoints.

```python
from fastapi import Page, new_pagination

Paginate = new_pagination(min_page_size=5, max_page_size=500)

@app.get("/heros", response_model=Page[HeroModel])
async def get_heros(paginate:Paginate):
    return paginate(select(Hero))
```

--------------------------------

### SQLAlchemy Async Session Dependency Injection

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Demonstrates how to use FastSQLAlchemy's `Session` as a dependency in FastAPI route handlers. The session is automatically managed, committed on success, and rolled back on failure.

```python
from fastsqla import Session
from sqlalchemy import select

@app.get("/heros")
async def get_heros(session:Session):
    stmt = select(...)
    result = await session.execute(stmt)
    ...
```

--------------------------------

### FastAPI Endpoint with Basic Pagination

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Demonstrates basic pagination using the `Paginate` dependency in a FastAPI endpoint. It injects a callable that accepts a SQLAlchemy `Select` and returns a `Page` model. Default query parameters are `offset` and `limit`.

```python
from fastapi import FastAPI
from fastsqla import Base, Page, Paginate
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI()

class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    age: Mapped[int]

class HeroModel(BaseModel):
    id: int
    name: str
    age: int

@app.get("/heroes")
async def list_heroes(paginate: Paginate[HeroModel]) -> Page[HeroModel]:
    return await paginate(select(Hero))
```

--------------------------------

### FastSQLAlchemy Pagination Customization

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Allows customization of pagination parameters like minimum and maximum page sizes. This provides flexibility in controlling the pagination behavior of API endpoints.

```python
from fastapi import Page, new_pagination
from sqlalchemy import select

Paginate = new_pagination(min_page_size=5, max_page_size=500)

@app.get("/heros", response_model=Page[HeroModel])
async def get_heros(paginate:Paginate):
    return paginate(select(Hero))
```

--------------------------------

### Filtered Hero Count Query with FastAPI Dependencies

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Demonstrates how to create a count query that respects the same filters applied to the main query by leveraging FastAPI dependencies. This ensures the count accurately reflects the filtered results.

```python
from sqlalchemy import func, select
from fastsqla import Session

async def filtered_hero_count(
    session: Session,
    age: int | None = None,
    name: str | None = None,
) -> int:
    stmt = select(func.count()).select_from(Hero)
    if age is not None:
        stmt = stmt.where(Hero.age == age)
    if name is not None:
        stmt = stmt.where(Hero.name.ilike(f"%{name}%"))
    result = await session.execute(stmt)
    return cast(int, result.scalar())

FilteredPaginate = Annotated[
    PaginateType[HeroModel],
    Depends(new_pagination(query_count_dependency=filtered_hero_count)),
]

@app.get("/heroes")
async def list_heroes(
    paginate: FilteredPaginate,
    age: int | None = None,
    name: str | None = None,
) -> Page[HeroModel]:
    stmt = select(Hero)
    if age is not None:
        stmt = stmt.where(Hero.age == age)
    if name is not None:
        stmt = stmt.where(Hero.name.ilike(f"%{name}%"))
    return await paginate(stmt)
```

--------------------------------

### Open Session Context Manager

Source: https://context7.com/hadrien/fastsqla/llms.txt

The `open_session` context manager simplifies session handling outside of FastAPI endpoints, automatically managing commits, rollbacks, and closing.

```APIDOC
## open_session

### Description
The `open_session` async context manager provides sessions outside of FastAPI endpoints, such as in background tasks or CLI scripts. It handles commit/rollback/close automatically.

### Usage
```python
from fastsqla import open_session
from sqlalchemy import select, update

async def background_sync_job():
    """Background task that updates hero statistics."""
    async with open_session() as session:
        # Read data
        stmt = select(Hero).where(Hero.age > 100)
        result = await session.execute(stmt)
        ancient_heroes = result.scalars().all()

        # Update data
        for hero in ancient_heroes:
            hero.status = "legendary"

        # Session auto-commits on context exit

async def batch_update_ages(hero_ids: list[int], new_age: int):
    """Batch update multiple heroes."""
    async with open_session() as session:
        stmt = update(Hero).where(Hero.id.in_(hero_ids)).values(age=new_age)
        await session.execute(stmt)
        # Auto-commits on success, auto-rollback on exception
```
```

--------------------------------

### Lifespan Management

Source: https://context7.com/hadrien/fastsqla/llms.txt

Manages the database connection lifecycle at application startup using environment variables.

```APIDOC
## Lifespan Management

### Description
The `lifespan` context manager sets up SQLAlchemy from environment variables at application startup. It configures the async engine, prepares ORM models, and manages connection pool lifecycle automatically.

### Usage
```python
from fastapi import FastAPI
from fastsqla import lifespan

# Basic setup - reads SQLALCHEMY_URL from environment
app = FastAPI(lifespan=lifespan)

# Run with environment variable:
# SQLALCHEMY_URL=postgresql+asyncpg://user:pass@localhost/db uvicorn app:app
```
```

--------------------------------

### Custom Lifespan with Programmatic Configuration

Source: https://context7.com/hadrien/fastsqla/llms.txt

Illustrates creating a custom lifespan context manager using `new_lifespan` for programmatic database configuration. This allows passing specific engine parameters like connection URL, pool size, and echo directly in the code, bypassing environment variables.

```python
from fastapi import FastAPI
from fastsqla import new_lifespan

# Programmatic configuration with custom parameters
lifespan = new_lifespan(
    "postgresql+asyncpg://user:password@localhost/mydb",
    pool_size=20,
    pool_recycle=1800,
    echo=True
)

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Adding Filters to Pagination

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Shows how to combine the `Paginate` dependency with additional query parameters for filtering results.

```APIDOC
## Adding Filters

Combine `Paginate` with additional query parameters:

### Endpoint
`GET /heroes`

### Query Parameters
- `age` (int, optional) - Filter by hero's age.
- `name` (str, optional) - Filter by hero's name (case-insensitive partial match).

### Code Example
```python
from fastapi import FastAPI
from fastsqla import Base, Page, Paginate
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI()

# ... (Hero and HeroModel definitions as above) ...

@app.get("/heroes")
async def list_heroes(
    paginate: Paginate[HeroModel],
    age: int | None = None,
    name: str | None = None,
):
    stmt = select(Hero)
    if age is not None:
        stmt = stmt.where(Hero.age == age)
    if name is not None:
        stmt = stmt.where(Hero.name.ilike(f"%{name}%"))
    return await paginate(stmt)
```

### Request Example
`GET /heroes?age=30&name=pond`

### Response Example (Success 200)
```json
{
  "data": [
    { "id": 1, "name": "Deadpond", "age": 30 }
  ],
  "meta": {
    "offset": 0,
    "total_items": 1,
    "total_pages": 1,
    "page_number": 1
  }
}
```
```

--------------------------------

### Custom Count Query Dependency for Pagination

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Demonstrates how to provide a custom count query dependency to `new_pagination()` for scenarios where the default `SELECT COUNT(*)` is inefficient. This custom dependency receives the session and returns the total item count.

```python
from typing import cast
from sqlalchemy import func, select
from fastsqla import Session

# Assuming 'Sticky' and 'StickyModel' are defined elsewhere
# async def query_count(session: Session) -> int:
#     result = await session.execute(select(func.count()).select_from(Sticky))
#     return cast(int, result.scalar())

# CustomPaginate = Annotated[
#     PaginateType[StickyModel],
#     Depends(new_pagination(query_count_dependency=query_count)),
# ]
```

--------------------------------

### Custom Result Processors for Multi-Column Selects

Source: https://github.com/hadrien/fastsqla/blob/main/skills/fastsqla-pagination/SKILL.md

Explains how to customize the result processor for different select scenarios. The default processor is suitable for single-entity selects, while `.mappings()` is recommended for multi-column selects or joins.

```python
lambda result: iter(result.unique().scalars())
```

```python
lambda result: iter(result.mappings())
```

--------------------------------

### FastSQLAlchemy Built-in Pagination

Source: https://github.com/hadrien/fastsqla/blob/main/docs/index.md

Implements automatic pagination for API endpoints using FastSQLAlchemy's `Page` and `Paginate` components. It handles offset and limit parameters, returning paginated data with metadata.

```python
from fastsqla import Page, Paginate
from sqlalchemy import select

@app.get("/heros", response_model=Page[HeroModel])
async def get_heros(paginate:Paginate):
    return await paginate(select(Hero))
```