Quick Start

This guide will get you up and running with the SQLSpec Litestar extension in 5 minutes.

Overview

In this quickstart, you’ll:

  1. Install SQLSpec with Litestar support

  2. Configure a database connection

  3. Create a Litestar application with the SQLSpec plugin

  4. Use dependency injection to access the database

  5. Execute queries in route handlers

Prerequisites

Ensure you have installed:

  • SQLSpec with a database adapter (see Installation)

  • Litestar web framework

pip install sqlspec[asyncpg,litestar]

Step 1: Import Required Modules

Step 2: Configure Database

Create a SQLSpec instance and add a database configuration:

spec = SQLSpec()
db = spec.add_config(
    AsyncpgConfig(
        pool_config={
            "dsn": "postgresql://user:password@localhost:5432/mydb",
            "min_size": 5,
            "max_size": 20
        },
        extension_config={
            "litestar": {
                "commit_mode": "autocommit"
            }
        }
    )
)

Note

Connection strings vary by database. See Dependency Injection for examples for each database.

For local development with SQLite:

from sqlspec.adapters.aiosqlite import AiosqliteConfig

db = spec.add_config(
    AiosqliteConfig(
        pool_config={"database": "./myapp.db"},
        extension_config={
            "litestar": {"commit_mode": "autocommit"}
        }
    )
)

Step 3: Create Route Handlers

Define route handlers that use dependency injection to access the database:

@get("/users")
async def list_users(db_session: AsyncpgDriver) -> dict:
    result = await db_session.execute("SELECT * FROM users LIMIT 10")
    return {"users": result.all()}

@get("/users/{user_id:int}")
async def get_user(
    user_id: int,
    db_session: AsyncpgDriver
) -> dict:
    result = await db_session.execute(
        "SELECT * FROM users WHERE id = $1",
        user_id
    )
    return result.one()

@post("/users")
async def create_user(
    data: dict,
    db_session: AsyncpgDriver
) -> dict:
    result = await db_session.execute(
        "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id, name, email",
        data["name"],
        data["email"]
    )
    return result.one()

Step 4: Create the Litestar App

Register the SQLSpec plugin with your Litestar application:

plugin = SQLSpecPlugin(sqlspec=spec)

app = Litestar(
    route_handlers=[list_users, get_user, create_user],
    plugins=[plugin]
)

Tip

The plugin automatically handles database lifecycle management including connection pooling, transaction management, and graceful shutdown.

Step 5: Run the Application

Run your Litestar application:

litestar run

You should see output similar to:

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Complete Example

Here’s a complete working example:

from litestar import Litestar, get, post
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig, AsyncpgDriver
from sqlspec.extensions.litestar import SQLSpecPlugin

# Configure database
spec = SQLSpec()
db = spec.add_config(
    AsyncpgConfig(
        pool_config={
            "dsn": "postgresql://user:password@localhost:5432/mydb",
            "min_size": 5,
            "max_size": 20
        },
        extension_config={
            "litestar": {"commit_mode": "autocommit"}
        }
    )
)

# Route handlers
@get("/users")
async def list_users(db_session: AsyncpgDriver) -> dict:
    result = await db_session.execute(
        "SELECT id, name, email FROM users ORDER BY id LIMIT 10"
    )
    return {"users": result.all()}

@get("/users/{user_id:int}")
async def get_user(
    user_id: int,
    db_session: AsyncpgDriver
) -> dict:
    result = await db_session.execute(
        "SELECT id, name, email FROM users WHERE id = $1",
        user_id
    )
    return result.one()

@post("/users")
async def create_user(
    data: dict,
    db_session: AsyncpgDriver
) -> dict:
    result = await db_session.execute(
        "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id, name, email",
        data["name"],
        data["email"]
    )
    return result.one()

# Create Litestar app
plugin = SQLSpecPlugin(sqlspec=spec)
app = Litestar(
    route_handlers=[list_users, get_user, create_user],
    plugins=[plugin]
)

Testing the API

Once your application is running, test the endpoints:

# List users
curl http://localhost:8000/users

# Get specific user
curl http://localhost:8000/users/1

# Create user
curl -X POST http://localhost:8000/users \
     -H "Content-Type: application/json" \
     -d '{"name": "Alice", "email": "[email protected]"}'

Type-Safe Results

For type-safe results, define Pydantic models:

from pydantic import BaseModel

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

@get("/users/{user_id:int}")
async def get_user(
    user_id: int,
    db_session: AsyncpgDriver
) -> User:
    result = await db_session.execute(
        "SELECT id, name, email FROM users WHERE id = $1",
        user_id,
        schema_type=User
    )
    return result.one()

Your IDE provides autocomplete and type checking for the returned user!

Database Setup

Create the users table:

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

You can use Litestar CLI to manage migrations:

# Generate migration
litestar db migrations generate -m "create users table"

# Apply migrations
litestar db migrations upgrade

# Rollback migration (if needed)
litestar db migrations downgrade

Commit Modes

The extension supports three transaction commit modes:

Manual Mode

Explicit transaction control (default):

db = spec.add_config(
    AsyncpgConfig(
        pool_config={"dsn": "postgresql://..."},
        extension_config={"litestar": {"commit_mode": "manual"}}
    )
)

@post("/users")
async def create_user(
    data: dict,
    db_session: AsyncpgDriver
) -> dict:
    async with db_session.begin_transaction():
        result = await db_session.execute(
            "INSERT INTO users (name) VALUES ($1) RETURNING id",
            data["name"]
        )
        return result.one()

Autocommit Mode

Automatic commit on 2XX responses (recommended):

db = spec.add_config(
    AsyncpgConfig(
        pool_config={"dsn": "postgresql://..."},
        extension_config={"litestar": {"commit_mode": "autocommit"}}
    )
)

@post("/users")
async def create_user(
    data: dict,
    db_session: AsyncpgDriver
) -> dict:
    # Automatically commits on success (2XX response)
    # Automatically rolls back on error (4XX/5XX response)
    result = await db_session.execute(
        "INSERT INTO users (name) VALUES ($1) RETURNING id",
        data["name"]
    )
    return result.one()

Autocommit with Redirects

Commits on both 2XX and 3XX responses:

db = spec.add_config(
    AsyncpgConfig(
        pool_config={"dsn": "postgresql://..."},
        extension_config={
            "litestar": {"commit_mode": "autocommit_include_redirect"}
        }
    )
)

Next Steps

To learn more:

Common Patterns

Health Check Endpoint

from sqlspec.adapters.asyncpg import AsyncpgConnection

@get("/health")
async def health_check(db_connection: AsyncpgConnection) -> dict:
    try:
        await db_connection.fetchval("SELECT 1")
        return {"status": "healthy", "database": "connected"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

Error Handling

from litestar import HTTPException
from litestar.status_codes import HTTP_404_NOT_FOUND

@get("/users/{user_id:int}")
async def get_user(
    user_id: int,
    db_session: AsyncpgDriver
) -> dict:
    result = await db_session.execute(
        "SELECT * FROM users WHERE id = $1",
        user_id
    )
    user = result.one_or_none()
    if not user:
        raise HTTPException(
            status_code=HTTP_404_NOT_FOUND,
            detail=f"User {user_id} not found"
        )
    return user

See Also