Litestar

https://docs.litestar.dev/latest
Admin
Litestar is a powerful, flexible, and highly performant ASGI framework that includes dependency...
Tokens:436,729
Snippets:4,398
Trust Score:8.4
Update:1 month ago
Show doc for...
Tokens:
Raw
### Full SQLAlchemy Integration Example

Source: https://docs.litestar.dev/latest/tutorials/repository-tutorial/04-repository-other.html

This snippet shows a complete Litestar application setup using the SQLAlchemy plugin. It includes database initialization, route handlers for blog posts, and repository dependency injection.

```python
sqlalchemy_config = SQLAlchemyConfig(connection_string="sqlite:///./db.sqlite")


class UUIDAuditBase(Base):
    __abstract__ = True
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow)


class BlogPost(UUIDAuditBase):
    __tablename__ = "blog_posts"

    title: Mapped[str] = mapped_column(String(100), unique=True)
    slug: Mapped[str] = mapped_column(String(100), unique=True)
    content: Mapped[str] = mapped_column(String(1000))


class BlogPostDTO(BaseModel):
    id: uuid.UUID
    title: str
    slug: str
    content: str


class BlogPostCreate(BaseModel):
    title: str
    content: str


class BlogPostRepository(SQLAlchemyRepository[BlogPost]):
    model = BlogPost

    async def get_available_slug(self, title: str) -> str:
        slug = slugify(title)
        if await self.exists(slug=slug):
            slug = f"{slug}-{uuid.uuid4().hex[:8]}"
        return slug


def provide_blog_post_repo(db_session: Session) -> BlogPostRepository:
    return BlogPostRepository(session=db_session)


sqlalchemy_plugin = SQLAlchemyInitPlugin(config=sqlalchemy_config)


async def on_startup() -> None:
    """Initializes the database."""
    async with sqlalchemy_config.get_engine().begin() as conn:
        await conn.run_sync(UUIDAuditBase.metadata.create_all)


@get(path="/")
async def get_blogs(
    blog_post_repo: BlogPostRepository,
) -> list[BlogPostDTO]:
    """Interact with SQLAlchemy engine and session."""
    objs = await blog_post_repo.list()
    type_adapter = TypeAdapter(list[BlogPostDTO])
    return type_adapter.validate_python(objs)


@get(path="/{post_slug:str}")
async def get_blog_details(
    post_slug: str,
    blog_post_repo: BlogPostRepository,
) -> BlogPostDTO:
    """Interact with SQLAlchemy engine and session."""
    obj = await blog_post_repo.get_one(slug=post_slug)
    return BlogPostDTO.model_validate(obj)


@post(path="/")
async def create_blog(
    blog_post_repo: BlogPostRepository,
    data: BlogPostCreate,
) -> BlogPostDTO:
    """Create a new blog post."""
    _data = data.model_dump(exclude_unset=True, by_alias=False, exclude_none=True)
    _data["slug"] = await blog_post_repo.get_available_slug(_data["title"])
    obj = await blog_post_repo.add(BlogPost(**_data))
    await blog_post_repo.session.commit()
    return BlogPostDTO.model_validate(obj)


app = Litestar(
    route_handlers=[create_blog, get_blogs, get_blog_details],
    dependencies={"blog_post_repo": Provide(provide_blog_post_repo, sync_to_thread=False)},
    on_startup=[on_startup],
    plugins=[SQLAlchemyInitPlugin(config=sqlalchemy_config)],
)
```

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

### Example Alias File for Application Management

Source: https://docs.litestar.dev/latest/topics/deployment/supervisor.html

A bash script that defines aliases for managing an application through Supervisor. This script allows for simplified commands like 'exampleapp start' instead of 'sudo supervisorctl start exampleapp'.

```bash
 1exampleapp() {
 2  case $1 in
 3    start)
 4      echo "Starting exampleapp..."
 5      sudo supervisorctl start exampleapp
 6      ;; 
 7 
 8    stop)
 9      echo "Stopping exampleapp..."
10      sudo supervisorctl stop exampleapp
11      ;; 
12 
13    restart)
14      echo "Restarting exampleapp..."
15      sudo supervisorctl restart exampleapp
16      ;; 
17 
18    status)
19      echo "Checking status of exampleapp..."
20      sudo supervisorctl status exampleapp
21      ;; 
22 
23    watch)
24      echo "Tailing logs for exampleapp..."
25      sudo supervisorctl tail -f exampleapp
26      ;; 
27 
28    help)
29      cat << EOF
30      Available options:
31        exampleapp start    - Start the exampleapp service
32        exampleapp stop     - Stop the exampleapp service
33        exampleapp restart  - Restart the exampleapp service
34        exampleapp status   - Check the status of the exampleapp service
35        exampleapp watch    - Tail the logs for the exampleapp service
36      EOF
37      ;; 
38 
39    *) 
40      echo "Unknown command: $1"
41      echo "Use 'exampleapp help' for a list of available commands."
42      ;; 
43  esac
44}

```

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

### Subprocess Test Client Setup

Source: https://docs.litestar.dev/latest/release-notes/changelog.html

Illustrates how to use `subprocess_async_client` and `subprocess_sync_client` for end-to-end testing by running an application in a separate process. Ensure `uvicorn` is installed.

```python
# Example usage would go here, but is not provided in the source text.
# The source text only mentions the existence of these functions.
pass

```

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

### Complete SQLAlchemy and Litestar Setup

Source: https://docs.litestar.dev/latest/tutorials/sqlalchemy/3-init-plugin.html

This code provides a full example of setting up a Litestar application with SQLAlchemy, including model definitions, dependency injection for database transactions, and API routes for CRUD operations.

```python
from collections.abc import AsyncGenerator

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, get, post, put
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import (
    SQLAlchemyAsyncConfig,
    SQLAlchemyInitPlugin,
    SQLAlchemySerializationPlugin,
)
from litestar.status_codes import HTTP_409_CONFLICT


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_items"

    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


async def provide_transaction(db_session: AsyncSession) -> AsyncGenerator[AsyncSession, None]:
    try:
        async with db_session.begin():
            yield db_session
    except IntegrityError as exc:
        raise ClientException(
            status_code=HTTP_409_CONFLICT,
            detail=str(exc),
        ) from exc


def get_todo_by_title(todo_name: FromPath[str], session: AsyncSession) -> TodoItem:
    query = select(TodoItem).where(TodoItem.title == todo_name)
    result = await session.execute(query)
    try:
        return result.scalar_one()
    except NoResultFound as e:
        raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e


async def get_todo_list(done: FromQuery[bool | None], session: AsyncSession) -> list[TodoItem]:
    query = select(TodoItem)
    if done is not None:
        query = query.where(TodoItem.done.is_(done))

    result = await session.execute(query)
    return list(result.scalars().all())


@get("/")
async def get_list(transaction: AsyncSession, done: FromQuery[bool | None] = None) -> list[TodoItem]:
    return await get_todo_list(done, transaction)


@post("/")
async def add_item(data: TodoItem, transaction: AsyncSession) -> TodoItem:
    transaction.add(data)
    return data


@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoItem, transaction: AsyncSession) -> TodoItem:
    todo_item = await get_todo_by_title(item_title, transaction)
    todo_item.title = data.title
    todo_item.done = data.done
    return todo_item


db_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///todo.sqlite", metadata=Base.metadata, create_all=True
)

app = Litestar(
    [get_list, add_item, update_item],
    dependencies={"transaction": provide_transaction},
    plugins=[
        SQLAlchemySerializationPlugin(),
        SQLAlchemyInitPlugin(db_config),
    ],
)
```

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

### InitPlugin Implementation Example (Python 3.8+)

Source: https://docs.litestar.dev/latest/usage/plugins/index.html

This example demonstrates a simple InitPlugin that adds a route handler and a dependency to the application configuration. It requires Python 3.8 or higher.

```python
from typing import Dict

from litestar import Litestar, get
from litestar.config.app import AppConfig
from litestar.di import NamedDependency, Provide
from litestar.plugins import InitPlugin


@get("/", sync_to_thread=False)
def route_handler(name: NamedDependency[str]) -> Dict[str, str]:
    return {"hello": name}


def get_name() -> str:
    return "world"


class MyPlugin(InitPlugin):
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        app_config.dependencies["name"] = Provide(get_name, sync_to_thread=False)
        app_config.route_handlers.append(route_handler)
        return app_config


app = Litestar(plugins=[MyPlugin()])

```

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

### Full Example: SQLAlchemy Session with DI

Source: https://docs.litestar.dev/latest/tutorials/sqlalchemy/1-provide-session-with-di.html

This snippet shows a complete Litestar application setup, including SQLAlchemy models, database connection management, a dependency provider for the session, and route handlers that utilize the injected session.

```python
1from contextlib import asynccontextmanager
2from typing import Any
3from collections.abc import AsyncGenerator
4
5from sqlalchemy import select
6from sqlalchemy.exc import IntegrityError, NoResultFound
7from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
8from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
9
10from litestar import Litestar, get, post, put
11from litestar.datastructures import State
12from litestar.di import NamedDependency
13from litestar.exceptions import ClientException, NotFoundException
14from litestar.params import FromPath, FromQuery
15from litestar.status_codes import HTTP_409_CONFLICT
16
17TodoType = dict[str, Any]
18TodoCollectionType = list[TodoType]
19
20
21class Base(DeclarativeBase): ...
22
23
24class TodoItem(Base):
25    __tablename__ = "todo_items"
26
27    title: Mapped[str] = mapped_column(primary_key=True)
28    done: Mapped[bool]
29
30
31@asynccontextmanager
32async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
33    engine = getattr(app.state, "engine", None)
34    if engine is None:
35        engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite")
36        app.state.engine = engine
37
38    async with engine.begin() as conn:
39        await conn.run_sync(Base.metadata.create_all)
40
41    try:
42        yield
43    finally:
44        await engine.dispose()
45
46
47sessionmaker = async_sessionmaker(expire_on_commit=False)
48
49
50async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
51    async with sessionmaker(bind=state.engine) as session:
52        try:
53            async with session.begin():
54                yield session
55        except IntegrityError as exc:
56            raise ClientException(
57                status_code=HTTP_409_CONFLICT,
58                detail=str(exc),
59            ) from exc
60
61
62def serialize_todo(todo: TodoItem) -> TodoType:
63    return {"title": todo.title, "done": todo.done}
64
65
66async def get_todo_by_title(todo_name: FromPath[str], session: AsyncSession) -> TodoItem:
67    query = select(TodoItem).where(TodoItem.title == todo_name)
68    result = await session.execute(query)
69    try:
70        return result.scalar_one()
71    except NoResultFound as e:
72        raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e
73
74
75async def get_todo_list(done: FromQuery[bool | None], session: AsyncSession) -> list[TodoItem]:
76    query = select(TodoItem)
77    if done is not None:
78        query = query.where(TodoItem.done.is_(done))
79
80    result = await session.execute(query)
81    return result.scalars().all()
82
83
84@get("/")
85async def get_list(
86    transaction: NamedDependency[AsyncSession], done: FromQuery[bool | None] = None
87) -> TodoCollectionType:
88    return [serialize_todo(todo) for todo in await get_todo_list(done, transaction)]
89
90
91@post("/")
92async def add_item(data: TodoType, transaction: NamedDependency[AsyncSession]) -> TodoType:
93    new_todo = TodoItem(title=data["title"], done=data["done"])
94    transaction.add(new_todo)
95    return serialize_todo(new_todo)
96
97
98@put("/{item_title:str}")
99async def update_item(
100    item_title: FromPath[str], data: TodoType, transaction: NamedDependency[AsyncSession]
101) -> TodoType:
102    todo_item = await get_todo_by_title(item_title, transaction)
103    todo_item.title = data["title"]
104    todo_item.done = data["done"]
105    return serialize_todo(todo_item)
106
107
108app = Litestar(
109    [get_list, add_item, update_item],
110    dependencies={"transaction": provide_transaction},
111    lifespan=[db_connection],
112)
```

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

### Client-Side Session Example

Source: https://docs.litestar.dev/latest/usage/middleware/builtin-middleware.html

Demonstrates setting up and using client-side sessions with cookie backend configuration. This example shows how to check, create, and delete session data.

```Python
from os import urandom

from litestar import Litestar, Request, delete, get, post
from litestar.middleware.session.client_side import CookieBackendConfig

# we initialize to config with a 16 byte key, i.e. 128 a bit key.
# in real world usage we should inject the secret from the environment
session_config = CookieBackendConfig(secret=urandom(16))  # type: ignore[arg-type]


@get("/session", sync_to_thread=False)
def check_session_handler(request: Request) -> dict[str, bool]:
    """Handler function that accesses request.session."""
    return {"has_session": request.session != {}}


@post("/session", sync_to_thread=False)
def create_session_handler(request: Request) -> None:
    """Handler to set the session."""
    if not request.session:
        # value can be a dictionary or pydantic model
        request.set_session({"username": "moishezuchmir"})


@delete("/session", sync_to_thread=False)
def delete_session_handler(request: Request) -> None:
    """Handler to clear the session."""
    if request.session:
        request.clear_session()


app = Litestar(
    route_handlers=[check_session_handler, create_session_handler, delete_session_handler],
    middleware=[session_config.middleware],
)
```

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

### PsycoPgChannelsBackend Startup

Source: https://docs.litestar.dev/latest/reference/channels/backends/psycopg.html

Called by the plugin on application startup. This method handles any necessary setup for the backend when the Litestar application starts.

```python
_async _on_startup() -> None:
```

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

### Litestar Test Client Example

Source: https://docs.litestar.dev/latest/onboarding/flask.html

Shows how to use Litestar's create_test_client to test a simple GET route. The client is used within a context manager.

```python
from litestar import get
from litestar.status_codes import HTTP_200_OK
from litestar.testing import create_test_client


@get("/", sync_to_thread=False)
def index() -> str:
    return "Index Page"


def test_index() -> None:
    with create_test_client(route_handlers=[index]) as client:
        response = client.get("/")
    assert response.status_code == HTTP_200_OK
    assert response.text == "Index Page"

```

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

### SQLAlchemy Sync Plugin Example (Python 3.8+)

Source: https://docs.litestar.dev/latest/usage/databases/sqlalchemy/plugins/sqlalchemy_plugin.html

Illustrates the synchronous SQLAlchemy plugin setup for Python 3.8+, including model definition and a sync route handler that runs in a thread.

```python
from __future__ import annotations

from typing import TYPE_CHECKING, Sequence

from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.di import NamedDependency
from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemySyncConfig

if TYPE_CHECKING:
    from sqlalchemy.orm import Session


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


@post("/", sync_to_thread=True)
def add_item(data: TodoItem, db_session: NamedDependency[Session]) -> Sequence[TodoItem]:
    with db_session.begin():
        db_session.add(data)
    return db_session.execute(select(TodoItem)).scalars().all()


config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite", create_all=True, metadata=Base.metadata)
plugin = SQLAlchemyPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin])

```

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

### Flask Test Client Example

Source: https://docs.litestar.dev/latest/onboarding/flask.html

Demonstrates how to use Flask's test client to make a GET request and assert the response status code and text.

```python
def test_index() -> None:
    client = app.test_client()
    response = client.get("/")
    assert response.status_code == 200
    assert response.text == "Index Page"

```

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

### Defining and Registering a User Order Controller

Source: https://docs.litestar.dev/latest/usage/routing/overview.html

This example demonstrates defining a controller for user orders with various HTTP methods (POST, GET, PATCH, DELETE). It shows how the controller's path is combined with handler paths and how to use DTOs for request/response handling.

```python
from litestar.plugins.pydantic import PydanticDTO
from litestar.controller import Controller
from litestar.dto import DTOConfig, DTOData
from litestar.handlers import get, post, patch, delete
from pydantic import BaseModel, UUID4


class UserOrder(BaseModel):
   user_id: int
   order: str


class PartialUserOrderDTO(PydanticDTO[UserOrder]):
   config = DTOConfig(partial=True)


class UserOrderController(Controller):
   path = "/user-order"

   @post()
   async def create_user_order(self, data: UserOrder) -> UserOrder: ...

   @get(path="/{order_id:uuid}")
   async def retrieve_user_order(self, order_id: UUID4) -> UserOrder: ...

   @patch(path="/{order_id:uuid}", dto=PartialUserOrderDTO)
   async def update_user_order(
       self, order_id: UUID4, data: DTOData[PartialUserOrderDTO]
   ) -> UserOrder: ...

   @delete(path="/{order_id:uuid}")
   async def delete_user_order(self, order_id: UUID4) -> None: ...
```

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

### SQLAlchemy Plugin Configuration Example

Source: https://docs.litestar.dev/latest/tutorials/sqlalchemy/4-final-touches-and-recap.html

Example of configuring the SQLAlchemyPlugin with database configuration and dependency injection for transactions.

```python
15    [get_list, add_item, update_item],
16    dependencies={"transaction": provide_transaction},
17    plugins=[SQLAlchemyPlugin(db_config)],
```

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

### Install MiniJinja Extra

Source: https://docs.litestar.dev/latest/usage/templating.html

Install the MiniJinja template engine using pip with the 'minijinja' extra.

```bash
pip install 'litestar[minijinja]'
```

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

### Install Hypercorn

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Install the Hypercorn ASGI server using pip. Hypercorn supports HTTP/1.1, HTTP/2, and WebSocket.

```bash
pip install hypercorn
```

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

### Litestar Basic Setup with State

Source: https://docs.litestar.dev/latest/onboarding/django.html

Demonstrates how to set up a Litestar application with state management, including a custom client class.

```python
from litestar import Litestar, get
from litestar.datastructures import State


class RedisClient:
    """Stand-in for a configured Redis client."""


@get("/")
async def handler(state: State) -> dict[str, str]:
    redis: RedisClient = state.redis
    return {"client": type(redis).__name__}


app = Litestar(
    route_handlers=[handler],
    state=State({"redis": RedisClient()}),
)

```

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

### Full Example: User DTO with ClassicPagination

Source: https://docs.litestar.dev/latest/usage/dto/1-abstract-dto.html

This snippet demonstrates a complete example including SQLAlchemy model, User DTO with excluded fields, a GET endpoint returning paginated users, and the Litestar application setup.

```python
from datetime import datetime

from sqlalchemy.orm import Mapped

from litestar import Litestar, get
from litestar.dto import DTOConfig
from litestar.pagination import ClassicPagination
from litestar.plugins.sqlalchemy import SQLAlchemyDTO

from .my_lib import Base


class User(Base):
    name: Mapped[str]
    password: Mapped[str]
    created_at: Mapped[datetime]


class UserDTO(SQLAlchemyDTO[User]):
    config = DTOConfig(exclude={"password", "created_at"})


@get("/users", dto=UserDTO, sync_to_thread=False)
def get_users() -> ClassicPagination[User]:
    return ClassicPagination(
        page_size=10,
        total_pages=1,
        current_page=1,
        items=[
            User(
                id=1,
                name="Litestar User",
                password="xyz",
                created_at=datetime.now(),
            ),
        ],
    )


app = Litestar(route_handlers=[get_users])

```

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

### Enabling CSRF Protection with CSRFConfig

Source: https://docs.litestar.dev/latest/usage/middleware/builtin-middleware.html

To enable CSRF protection, pass an instance of CSRFConfig to the Litestar constructor. This example shows basic setup with safe (GET) and unsafe (POST) routes.

```python
from litestar import Litestar, get, post
from litestar.config.csrf import CSRFConfig


@get()
async def get_resource() -> str:
    # GET is one of the safe methods
    return "some_resource"

@post("{id:int}")
async def create_resource(id: int) -> bool:
    # POST is one of the unsafe methods
    return True

csrf_config = CSRFConfig(secret="my-secret")

app = Litestar([get_resource, create_resource], csrf_config=csrf_config)

```

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

### ValkeyStore.__init__()

Source: https://docs.litestar.dev/latest/reference/stores/index.html

Initializes a ValkeyStore instance.

```APIDOC
## ValkeyStore.__init__()

### Description
Initializes a ValkeyStore instance.

### Method
N/A (Constructor)

### Endpoint
N/A

### Parameters
#### Path Parameters
N/A

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
N/A

### Response
#### Success Response (200)
N/A

#### Response Example
N/A
```

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

### Litestar Route Handler Example

Source: https://docs.litestar.dev/latest/onboarding/fastapi.html

Illustrates a Litestar application with a root GET endpoint defined using the @get decorator.

```python
from litestar import Litestar, get


@get("/")
async def index() -> dict[str, str]:
    return {"hello": "world"}


app = Litestar(route_handlers=[index])
```

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

### Async Test Client to Get Session Data

Source: https://docs.litestar.dev/latest/reference/testing.html

Illustrates retrieving session data using AsyncTestClient after setting it. This example uses a GET route to access the session.

```python
from litestar import Litestar, get
from litestar.middleware.session.memory_backend import MemoryBackendConfig

session_config = MemoryBackendConfig()


@get(path="/test")
def get_session_data(request: Request) -> Dict[str, Any]:
    return request.session


app = Litestar(
    route_handlers=[get_session_data], middleware=[session_config.middleware]
)

async with AsyncTestClient(app=app, session_config=session_config) as client:
    await client.set_session_data({"foo": "bar"})
    assert await client.get("/test").json() == {"foo": "bar"}
```

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

### Serve Documentation Locally

Source: https://docs.litestar.dev/latest/contribution-guide.html

Serve the documentation locally using the make command. This is useful for previewing changes.

```bash
make docs-serve

```

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

### Create a Hello World App (Python 3.8+)

Source: https://docs.litestar.dev/latest/usage/applications.html

Instantiate the `Litestar` class with a list of route handlers. This example uses type hinting for Python 3.8.

```python
from typing import Dict

from litestar import Litestar, get


@get("/")
async def hello_world() -> Dict[str, str]:
    """Handler function that returns a greeting dictionary."""
    return {"hello": "world"}


app = Litestar(route_handlers=[hello_world])

```

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

### JWT Cookie Auth Setup and Usage (Python 3.9+)

Source: https://docs.litestar.dev/latest/usage/security/jwt.html

Provides a similar setup for JWTCookieAuth as the Python 3.8+ example, but utilizes Python 3.9+ type hinting features.

```python
import secrets
from os import environ
from typing import Any, Optional
from uuid import UUID

from pydantic import BaseModel, EmailStr

from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import JWTCookieAuth, Token


# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB  etc.
class User(BaseModel):
    id: UUID
    name: str
    email: EmailStr


MOCK_DB: dict[str, User] = {}


# JWTCookieAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:

```

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

### Initialize and use MemoryStore

Source: https://docs.litestar.dev/latest/release-notes/whats-new-2.html

Demonstrates how to initialize a `MemoryStore` and perform basic get/set operations. Note that accessing a non-existent key returns `None`.

```python
from litestar.stores.memory import MemoryStore

store = MemoryStore()


async def main() -> None:
    value = await store.get("key")
    print(value)  # this will print 'None', as no store with this key has been defined yet

    await store.set("key", b"value")
    value = await store.get("key")
    print(value)
```

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

### FastAPI Route Handler Example

Source: https://docs.litestar.dev/latest/onboarding/fastapi.html

Defines a basic FastAPI application with a root GET endpoint.

```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def index() -> dict[str, str]: ...
```

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

### Example Application Commands

Source: https://docs.litestar.dev/latest/topics/deployment/supervisor.html

Common commands to manage an application via Supervisor aliases.

```bash
exampleapp start
exampleapp stop
exampleapp restart
exampleapp status
exampleapp watch

```

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

### Initialize ValkeyStore with a new Valkey client

Source: https://docs.litestar.dev/latest/reference/stores/valkey.html

Initialize ValkeyStore by creating a new Valkey client instance using a connection URL. Various connection parameters like database, port, username, and password can be provided. A virtual key namespace can also be specified.

```python
from litestar.stores.valkey import ValkeyStore


async def my_handler():
    store = ValkeyStore.with_client(
        url="valkey://localhost:6379",
        db=1,
        username="user",
        password="password",
        namespace="my-app"
    )
```

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

### Minimal Client-Side Session Configuration

Source: https://docs.litestar.dev/latest/usage/middleware/builtin-middleware.html

A minimal setup for client-side sessions using `CookieBackendConfig`. Ensure the `cryptography` library is installed for this to function.

```Python
from os import urandom

from litestar import Litestar
from litestar.middleware.session.client_side import CookieBackendConfig

session_config = CookieBackendConfig(secret=urandom(16))  # type: ignore

app = Litestar(middleware=[session_config.middleware])

```

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

### Run Daphne Server

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Start the Daphne ASGI server with the default configuration. Assumes your app is named 'app' and located in a file named 'app.py'.

```bash
daphne app:app
```

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

### Granian Console Output

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Example console output when Granian starts successfully, indicating the application is running and the address it's accessible at.

```text
[INFO] Starting granian
[INFO] Listening at: 127.0.0.1:8000
```

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

### Daphne Console Output

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Example console output when Daphne starts successfully, indicating the application is running and the address it's accessible at.

```text
INFO - 2023-11-12 23:31:51,571 - daphne.cli - cli - Starting server at tcp:port=8000:interface=127.0.0.1
INFO - 2023-11-12 23:31:51,572 - daphne.server - server - Listening on TCP address 127.0.0.1:8000
```

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

### ValkeyStore.__init__

Source: https://docs.litestar.dev/latest/reference/stores/valkey.html

Initializes a ValkeyStore instance, allowing configuration of the Valkey client, namespace, and client shutdown handling.

```APIDOC
## ValkeyStore.__init__

### Description
Initializes a ValkeyStore instance, allowing configuration of the Valkey client, namespace, and client shutdown handling.

### Parameters
* **valkey** (Valkey) - An `valkey.asyncio.Valkey` instance.
* **namespace** (str | None | Literal[_EmptyEnum.EMPTY]) - A key prefix to simulate a namespace in valkey. Defaults to `LITESTAR`. Can be explicitly disabled by passing `None`, which makes `delete_all()` unavailable.
* **handle_client_shutdown** (bool) - If `True`, handles the shutdown of the valkey instance automatically during the store’s lifespan. Set to `False` unless shutdown is handled externally.
```

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

### Hypercorn Console Output

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Example console output when Hypercorn starts successfully, indicating the application is running and the address it's accessible at.

```text
[2023-11-12 23:31:26 -0800] [16748] [INFO] Running on http://127.0.0.1:8000 (CTRL + C to quit)
```

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

### Run Granian Server

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Start the Granian ASGI server with the default configuration, specifying the application interface. Assumes your app is named 'app' and located in a file named 'app.py'.

```bash
granian --interface asgi app:app
```

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

### Using Application State with Startup, Middleware, Dependencies, and Handlers

Source: https://docs.litestar.dev/latest/usage/applications.html

Demonstrates how to set and access application state across different parts of the application lifecycle, including startup hooks, middleware, dependencies, and route handlers.

```python
import logging
from typing import TYPE_CHECKING, Any

from litestar import Litestar, Request, get
from litestar.datastructures import State
from litestar.di import NamedDependency, Provide

if TYPE_CHECKING:
    from litestar.types import ASGIApp, Receive, Scope, Send

logger = logging.getLogger(__name__)


def set_state_on_startup(app: Litestar) -> None:
    """Startup and shutdown hooks can receive `State` as a keyword arg."""
    app.state.value = "abc123"


def middleware_factory(*, app: "ASGIApp") -> "ASGIApp":
    """A middleware can access application state via `scope`."""

    async def my_middleware(scope: "Scope", receive: "Receive", send: "Send") -> None:
        state = Litestar.from_scope(scope).state
        logger.info("state value in middleware: %s", state.value)
        await app(scope, receive, send)

    return my_middleware


async def my_dependency(state: State) -> Any:
    """Dependencies can receive state via injection."""
    logger.info("state value in dependency: %s", state.value)


@get("/", dependencies={"dep": Provide(my_dependency)}, middleware=[middleware_factory], sync_to_thread=False)
def get_handler(state: State, request: Request, dep: NamedDependency[Any]) -> None:
    """Handlers can receive state via injection."""
    logger.info("state value in handler from `State`: %s", state.value)
    logger.info("state value in handler from `Request`: %s", request.app.state.value)


app = Litestar(route_handlers=[get_handler], on_startup=[set_state_on_startup])

```

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

### Uvicorn Console Output

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Example console output when Uvicorn starts successfully, indicating the application is running and the address it's accessible at.

```text
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
```

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

### Run Hypercorn Server

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Start the Hypercorn ASGI server with the default configuration. Assumes your app is named 'app' and located in a file named 'app.py'.

```bash
hypercorn app:app
```

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

### Initializing Application State with a Dictionary (Python 3.9+)

Source: https://docs.litestar.dev/latest/usage/applications.html

Demonstrates initializing application state with a dictionary for Python 3.9+, making the state accessible within route handlers. Uses type hinting for dictionary return type.

```python
from typing import Any

from litestar import Litestar, get
from litestar.datastructures import State


@get("/", sync_to_thread=False)
def handler(state: State) -> dict[str, Any]:
    return state.dict()


app = Litestar(route_handlers=[handler], state=State({"count": 100}))

```

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

### Client Request and Server Response Example

Source: https://docs.litestar.dev/latest/usage/dto/1-abstract-dto.html

Demonstrates a client request to the /users endpoint with JSON data and the corresponding server response. Shows how 'private' and 'read-only' fields are handled.

```bash
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"name":"Litestar User","password":"xyz","created_at":"2023-04-24T00:00:00Z"}
{"created_at":"0001-01-01T00:00:00","id":"9d62d0a6-8c14-40d3-93d1-f48bb306b792","name":"Litestar User"}


```

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

### JWTAuth Setup and Application Initialization (Python 3.9+)

Source: https://docs.litestar.dev/latest/usage/security/jwt.html

Demonstrates how to set up JWTAuth with a user retrieval handler, exclude specific routes from authentication, and initialize the Litestar app with the JWT middleware. Requires Python 3.9+.

```python
import secrets
from os import environ
from typing import Any, Optional
from uuid import UUID

from pydantic import BaseModel, EmailStr

from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import JWTAuth, Token


# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB  etc.
class User(BaseModel):
    id: UUID
    name: str
    email: EmailStr


MOCK_DB: dict[str, User] = {}


# JWTAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: Token, connection: "ASGIConnection[Any, Any, Any, Any]") -> Optional[User]:
    # logic here to retrieve the user instance
    return MOCK_DB.get(token.sub)


jwt_auth = JWTAuth[User](
    retrieve_user_handler=retrieve_user_handler,
    token_secret=environ.get("JWT_SECRET", secrets.token_hex()),
    # we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
    # and our openAPI docs.
    exclude=["/login", "/schema"],
)


# Given an instance of 'JWTAuth' we can create a login handler function:
@post("/login")
async def login_handler(data: User) -> Response[User]:
    MOCK_DB[str(data.id)] = data
    # you can do whatever you want to update the response instance here
    # e.g. response.set_cookie(...)
    return jwt_auth.login(identifier=str(data.id), token_extras={"email": data.email}, response_body=data)


# We also have some other routes, for example:
@get("/some-path", sync_to_thread=False)
def some_route_handler(request: "Request[User, Token, Any]") -> Any:
    # request.user is set to the instance of user returned by the middleware
    assert isinstance(request.user, User)
    # request.auth is the instance of 'litestar_jwt.Token' created from the data encoded in the auth header
    assert isinstance(request.auth, Token)
    # do stuff ...


# We create our OpenAPIConfig as usual - the JWT security scheme will be injected into it.
openapi_config = OpenAPIConfig(
    title="My API",
    version="1.0.0",
)

# We initialize the app instance and pass the jwt_auth 'on_app_init' handler to the constructor.
# The hook handler will inject the JWT middleware and openapi configuration into the app.
app = Litestar(
    route_handlers=[login_handler, some_route_handler],
    on_app_init=[jwt_auth.on_app_init],
    openapi_config=openapi_config,
)

```

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

### FileStore.__init__()

Source: https://docs.litestar.dev/latest/reference/stores/index.html

Initializes a FileStore instance.

```APIDOC
## FileStore.__init__()

### Description
Initializes a FileStore instance.

### Method
N/A (Constructor)

### Endpoint
N/A

### Parameters
#### Path Parameters
N/A

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
N/A

### Response
#### Success Response (200)
N/A

#### Response Example
N/A
```

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

### JWTAuth Setup and Application Initialization (Python 3.10+)

Source: https://docs.litestar.dev/latest/usage/security/jwt.html

Demonstrates how to set up JWTAuth with a user retrieval handler, exclude specific routes from authentication, and initialize the Litestar app with the JWT middleware. Requires Python 3.10+.

```python
import secrets
from os import environ
from typing import Any
from uuid import UUID

from pydantic import BaseModel, EmailStr

from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import JWTAuth, Token


# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB  etc.
class User(BaseModel):
    id: UUID
    name: str
    email: EmailStr


MOCK_DB: dict[str, User] = {}


# JWTAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: Token, connection: "ASGIConnection[Any, Any, Any, Any]") -> User | None:
    # logic here to retrieve the user instance
    return MOCK_DB.get(token.sub)


jwt_auth = JWTAuth[User](
    retrieve_user_handler=retrieve_user_handler,
    token_secret=environ.get("JWT_SECRET", secrets.token_hex()),
    # we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
    # and our openAPI docs.
    exclude=["/login", "/schema"],
)


# Given an instance of 'JWTAuth' we can create a login handler function:
@post("/login")
async def login_handler(data: User) -> Response[User]:
    MOCK_DB[str(data.id)] = data
    # you can do whatever you want to update the response instance here
    # e.g. response.set_cookie(...)
    return jwt_auth.login(identifier=str(data.id), token_extras={"email": data.email}, response_body=data)



```

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

### Litestar App with SSE Generator (Python 3.8+)

Source: https://docs.litestar.dev/latest/usage/testing.html

Defines a Litestar application with a GET endpoint that returns Server-Sent Events generated by an async generator. This setup is used for testing scenarios that require a live server.

```python
"""
Assemble components into an app that shall be tested
"""

from typing import AsyncGenerator

from litestar import Litestar, get
from litestar.response import ServerSentEvent
from litestar.types import SSEData


async def generator(topic: str) -> AsyncGenerator[SSEData, None]:
    count = 0
    while count < 2:
        yield topic
        count += 1


@get("/notify/{topic:str}")
async def get_notified(topic: str) -> ServerSentEvent:
    return ServerSentEvent(generator(topic), event_type="Notifier")


app = Litestar(route_handlers=[get_notified])

```

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

### Run Uvicorn Server

Source: https://docs.litestar.dev/latest/topics/deployment/manually-with-asgi-server.html

Start the Uvicorn ASGI server with the default configuration. Assumes your app is named 'app' and located in a file named 'app.py'.

```bash
uvicorn app:app
```

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

### Using the ASGIRouteHandler class directly

Source: https://docs.litestar.dev/latest/usage/routing/handlers.html

Shows an equivalent implementation to the @asgi() decorator example, but by directly instantiating and using the ASGIRouteHandler class. This handler also manages HTTP GET requests and provides error responses for unsupported requests.

```python
from litestar import Response
from litestar.handlers.asgi_handlers import ASGIRouteHandler
from litestar.status_codes import HTTP_400_BAD_REQUEST
from litestar.types import Scope, Receive, Send


@ASGIRouteHandler(path="/my-asgi-app")
async def my_asgi_app(scope: Scope, receive: Receive, send: Send) -> None:
   if scope["type"] == "http":
       if scope["method"] == "GET":
           response = Response({"hello": "world"})
           await response(scope=scope, receive=receive, send=send)
       return
   response = Response(
       {"detail": "unsupported request"}, status_code=HTTP_400_BAD_REQUEST
   )
   await response(scope=scope, receive=receive, send=send)
```

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

### RedisStore.__init__()

Source: https://docs.litestar.dev/latest/reference/stores/index.html

Initializes a RedisStore instance.

```APIDOC
## RedisStore.__init__()

### Description
Initializes a RedisStore instance.

### Method
N/A (Constructor)

### Endpoint
N/A

### Parameters
#### Path Parameters
N/A

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
N/A

### Response
#### Success Response (200)
N/A

#### Response Example
N/A
```

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

### Starting a Subscription with Context Management

Source: https://docs.litestar.dev/latest/reference/channels/plugin.html

The start_subscription method provides an asynchronous context manager for creating a Subscriber. Unsubscription is automatically handled upon exiting the context.

```python
async with plugin.start_subscription(channels="general") as subscriber:
    # Use the subscriber here
    async for event in subscriber.iter_events():
        print(event)

async with plugin.start_subscription(channels=["user::123", "notifications"], history=5) as subscriber:
    # Process events with history
    pass
```

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

### Example Plugin Usage

Source: https://docs.litestar.dev/latest/reference/plugins/problem_details.html

Demonstrates how to define and use a custom plugin with Litestar, including dependency injection and route registration.

```python
from litestar import Litestar, get
from litestar.di import Provide
from litestar.plugins import InitPluginProtocol


def get_name() -> str:
    return "world"


@get("/my-path")
def my_route_handler(name: str) -> dict[str, str]:
    return {"hello": name}


class MyPlugin(InitPluginProtocol):
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        app_config.dependencies["name"] = Provide(get_name)
        app_config.route_handlers.append(my_route_handler)
        return app_config


app = Litestar(plugins=[MyPlugin()])

```

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

### Litestar App with SSE Generator (Python 3.9+)

Source: https://docs.litestar.dev/latest/usage/testing.html

Defines a Litestar application with a GET endpoint that returns Server-Sent Events generated by an async generator. This setup is used for testing scenarios that require a live server. Uses `collections.abc.AsyncGenerator` for compatibility with Python 3.9+.

```python
"""
Assemble components into an app that shall be tested
"""

from collections.abc import AsyncGenerator

from litestar import Litestar, get
from litestar.response import ServerSentEvent
from litestar.types import SSEData


async def generator(topic: str) -> AsyncGenerator[SSEData, None]:
    count = 0
    while count < 2:
        yield topic
        count += 1


@get("/notify/{topic:str}")
async def get_notified(topic: str) -> ServerSentEvent:
    return ServerSentEvent(generator(topic), event_type="Notifier")


app = Litestar(route_handlers=[get_notified])

```

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

### Litestar App with SQLAlchemy (Python 3.9+)

Source: https://docs.litestar.dev/latest/tutorials/sqlalchemy/0-introduction.html

This snippet presents a Litestar application using SQLAlchemy, similar to the Python 3.8+ example but with updated type hinting syntax. It demonstrates the core components for a TODO list application, including database setup and API endpoint definitions.

```python
1from contextlib import asynccontextmanager
2from typing import Any, Optional
3from collections.abc import AsyncGenerator, Sequence
4
5from sqlalchemy import select
6from sqlalchemy.exc import IntegrityError, NoResultFound
7from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
8from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
9
10from litestar import Litestar, get, post, put
11from litestar.datastructures import State
12from litestar.exceptions import ClientException, NotFoundException
13from litestar.params import FromPath, FromQuery
14from litestar.status_codes import HTTP_409_CONFLICT
15
16TodoType = dict[str, Any]
17TodoCollectionType = list[TodoType]
18
19
20class Base(DeclarativeBase): ...
21
22
23class TodoItem(Base):
24    __tablename__ = "todo_items"
25
26    title: Mapped[str] = mapped_column(primary_key=True)
27    done: Mapped[bool]
28
```
Tokens:
Raw