tessl/pypi-litestar

Litestar is a powerful, flexible yet opinionated ASGI web framework specifically focused on building high-performance APIs.

—

Pending

Overview

Eval results

Files

HTTP Route Handlers

Name: tessl/pypi-litestar
Author: tessl

Decorators and classes for handling HTTP requests including GET, POST, PUT, DELETE, PATCH, and generic route handlers. Route handlers process incoming HTTP requests and generate responses.

Capabilities

HTTP Method Decorators

Decorators for creating route handlers for specific HTTP methods. Each decorator creates an HTTPRouteHandler instance configured for the specified HTTP method.

def get(
    path: str | Sequence[str] | None = None,
    *,
    after_request: AfterRequestHookHandler | None = None,
    after_response: AfterResponseHookHandler | None = None,
    background: BackgroundTask | BackgroundTasks | None = None,
    before_request: BeforeRequestHookHandler | None = None,
    cache: bool | int | CacheKeyBuilder = False,
    dependencies: Dependencies | None = None,
    dto: type[AbstractDTO] | None = None,
    etag: ETagConfig | None = None,
    exception_handlers: ExceptionHandlersMap | None = None,
    guards: Sequence[Guard] | None = None,
    http_method: HttpMethod | str = HttpMethod.GET,
    media_type: MediaType | str | None = None,
    middleware: Sequence[Middleware] | None = None,
    name: str | None = None,
    opt: dict[str, Any] | None = None,
    return_dto: type[AbstractDTO] | None = None,
    signature_namespace: dict[str, Any] | None = None,
    status_code: int | None = None,
    summary: str | None = None,
    description: str | None = None,
    tags: Sequence[str] | None = None,
    type_encoders: TypeEncodersMap | None = None,
    include_in_schema: bool = True,
    sync_to_thread: bool = True,
    content_encoding: str | None = None,
    content_media_type: str | None = None,
) -> Callable[[AnyCallable], HTTPRouteHandler]:
    """
    Create a route handler for GET requests.

    Parameters:
    - path: Route path(s) - can be a string or sequence of strings
    - after_request: Hook called after request processing
    - after_response: Hook called after response sending
    - background: Background task(s) to execute after response
    - before_request: Hook called before request processing
    - cache: Response caching configuration
    - dependencies: Route-specific dependency providers
    - dto: DTO for request body serialization
    - etag: ETag configuration for caching
    - exception_handlers: Route-specific exception handlers
    - guards: Authorization guards
    - http_method: HTTP method (defaults to GET)
    - media_type: Response media type
    - middleware: Route-specific middleware
    - name: Route name for URL generation
    - opt: Arbitrary options dictionary
    - return_dto: DTO for response serialization
    - signature_namespace: Additional namespace for signature inspection
    - status_code: Default response status code
    - summary: OpenAPI summary
    - description: OpenAPI description
    - tags: OpenAPI tags
    - type_encoders: Route-specific type encoders
    - include_in_schema: Include in OpenAPI schema
    - sync_to_thread: Run sync handlers in thread pool
    - content_encoding: Response content encoding
    - content_media_type: Response content media type

    Returns:
    Decorator function that creates an HTTPRouteHandler
    """

def post(
    path: str | Sequence[str] | None = None,
    **kwargs: Any
) -> Callable[[AnyCallable], HTTPRouteHandler]:
    """Create a route handler for POST requests."""

def put(
    path: str | Sequence[str] | None = None,
    **kwargs: Any
) -> Callable[[AnyCallable], HTTPRouteHandler]:
    """Create a route handler for PUT requests."""

def patch(
    path: str | Sequence[str] | None = None,
    **kwargs: Any
) -> Callable[[AnyCallable], HTTPRouteHandler]:
    """Create a route handler for PATCH requests."""

def delete(
    path: str | Sequence[str] | None = None,
    **kwargs: Any
) -> Callable[[AnyCallable], HTTPRouteHandler]:
    """Create a route handler for DELETE requests."""

def head(
    path: str | Sequence[str] | None = None,
    **kwargs: Any
) -> Callable[[AnyCallable], HTTPRouteHandler]:
    """Create a route handler for HEAD requests."""

def route(
    path: str | Sequence[str] | None = None,
    *,
    http_method: HttpMethod | str | Sequence[HttpMethod | str],
    **kwargs: Any
) -> Callable[[AnyCallable], HTTPRouteHandler]:
    """Create a route handler for multiple HTTP methods."""

HTTPRouteHandler Class

The underlying route handler class created by HTTP method decorators. Can be instantiated directly for advanced configuration.

class HTTPRouteHandler(BaseRouteHandler):
    def __init__(
        self,
        fn: AnyCallable,
        *,
        http_method: HttpMethod | str | Sequence[HttpMethod | str],
        path: str | Sequence[str] | None = None,
        status_code: int | None = None,
        media_type: MediaType | str | None = None,
        # ... same parameters as decorators
    ):
        """
        Create an HTTP route handler.

        Parameters:
        - fn: Handler function
        - http_method: HTTP method(s) to handle
        - path: Route path(s)
        - status_code: Default response status code
        - media_type: Response media type
        """

    @property
    def http_methods(self) -> set[Method]:
        """Get HTTP methods handled by this route."""
        
    async def handle_request(self, request: Request) -> Response:
        """Process incoming request and generate response."""

ASGI Route Handler

Route handler for integrating existing ASGI applications or middleware.

def asgi(
    path: str | Sequence[str] | None = None,
    *,
    is_mount: bool = False,
    **kwargs: Any
) -> Callable[[Callable[[Scope, Receive, Send], Awaitable[None]]], ASGIRouteHandler]:
    """
    Create an ASGI route handler.

    Parameters:
    - path: Route path(s)
    - is_mount: Whether to mount the ASGI app at the path
    """

class ASGIRouteHandler(BaseRouteHandler):
    def __init__(
        self,
        fn: Callable[[Scope, Receive, Send], Awaitable[None]],
        *,
        is_mount: bool = False,
        **kwargs: Any
    ):
        """Create an ASGI route handler."""

Path Parameters

Path parameter parsing with type conversion and validation.

# Path parameter types in route paths:
# {param_name:int} - Integer parameter  
# {param_name:float} - Float parameter
# {param_name:str} - String parameter (default)
# {param_name:uuid} - UUID parameter
# {param_name:path} - Path parameter (preserves slashes)

Usage Examples

Basic Route Handlers

from litestar import get, post, put, delete
from litestar.dto import DataclassDTO
from dataclasses import dataclass

@dataclass
class User:
    name: str
    email: str
    id: int | None = None

@get("/users")
async def list_users() -> list[User]:
    return [User(id=1, name="Alice", email="alice@example.com")]

@get("/users/{user_id:int}")
async def get_user(user_id: int) -> User:
    return User(id=user_id, name="Alice", email="alice@example.com")

@post("/users", dto=DataclassDTO[User])
async def create_user(data: User) -> User:
    # Simulate user creation
    data.id = 123
    return data

@put("/users/{user_id:int}", dto=DataclassDTO[User])
async def update_user(user_id: int, data: User) -> User:
    data.id = user_id
    return data

@delete("/users/{user_id:int}")
async def delete_user(user_id: int) -> dict[str, str]:
    return {"status": "deleted", "user_id": str(user_id)}

Path Parameters and Query Parameters

from litestar import get
from litestar.params import Parameter

@get("/users/{user_id:int}/posts/{post_id:uuid}")
async def get_user_post(
    user_id: int, 
    post_id: UUID,
    include_comments: bool = False,
    limit: int = Parameter(default=10, ge=1, le=100)
) -> dict:
    return {
        "user_id": user_id,
        "post_id": str(post_id),
        "include_comments": include_comments,
        "limit": limit
    }

Request Body Handling

from litestar import post
from litestar.dto import DataclassDTO
from dataclasses import dataclass
from typing import Optional

@dataclass
class CreateUserRequest:
    name: str
    email: str
    age: Optional[int] = None

@post("/users", dto=DataclassDTO[CreateUserRequest])
async def create_user_with_validation(data: CreateUserRequest) -> dict:
    # data is automatically validated and converted
    return {"message": f"Created user {data.name} with email {data.email}"}

# For raw request body access
@post("/upload")
async def upload_file(request: Request) -> dict:
    body = await request.body()
    return {"size": len(body)}

Multiple HTTP Methods

from litestar import route, HttpMethod

@route("/api/resource", http_method=[HttpMethod.GET, HttpMethod.POST])
async def handle_resource(request: Request) -> dict:
    if request.method == "GET":
        return {"action": "fetch"}
    elif request.method == "POST":
        data = await request.json()
        return {"action": "create", "data": data}

Response Configuration

from litestar import get, post
from litestar.response import Response
from litestar.status_codes import HTTP_201_CREATED, HTTP_202_ACCEPTED

@get("/health", status_code=200, media_type="text/plain")
async def health_check() -> str:
    return "OK"

@post("/async-task", status_code=HTTP_202_ACCEPTED)
async def start_async_task(data: dict) -> dict:
    # Start background processing
    return {"task_id": "12345", "status": "accepted"}

@post("/users", status_code=HTTP_201_CREATED)
async def create_user_with_status(data: dict) -> dict:
    # User creation logic
    return {"id": 123, "status": "created"}

Custom Response Types

from litestar import get
from litestar.response import File, Stream, Redirect, Template

@get("/download/{filename:str}")
async def download_file(filename: str) -> File:
    return File(f"/uploads/{filename}")

@get("/stream")
async def stream_data() -> Stream:
    async def generate():
        for i in range(100):
            yield f"data chunk {i}\n"
    
    return Stream(generate())

@get("/redirect")
async def redirect_example() -> Redirect:
    return Redirect("/new-location")

@get("/page")
async def render_page() -> Template:
    return Template("page.html", context={"title": "My Page"})

Error Handling in Routes

from litestar import get
from litestar.exceptions import HTTPException, NotFoundException
from litestar.status_codes import HTTP_400_BAD_REQUEST

@get("/users/{user_id:int}")
async def get_user_safe(user_id: int) -> dict:
    if user_id < 1:
        raise HTTPException(
            detail="User ID must be positive",
            status_code=HTTP_400_BAD_REQUEST
        )
    
    # Simulate user lookup
    if user_id == 999:
        raise NotFoundException("User not found")
    
    return {"id": user_id, "name": "Alice"}

Background Tasks

from litestar import post
from litestar.background_tasks import BackgroundTask

def send_email(email: str, message: str) -> None:
    # Email sending logic
    print(f"Sending email to {email}: {message}")

@post("/notify", background=BackgroundTask(send_email, "admin@example.com", "New user registered"))
async def notify_admin(user_data: dict) -> dict:
    return {"status": "success", "message": "Notification queued"}

Types

# Route handler function types
AnyCallable = Callable[..., Any]
RouteHandlerFunction = Callable[..., Any | Awaitable[Any]]

# HTTP method types
Method = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS", "TRACE"]

# Hook handler types
AfterRequestHookHandler = Callable[[Request], None | Awaitable[None]]
AfterResponseHookHandler = Callable[[Request, Response], None | Awaitable[None]]
BeforeRequestHookHandler = Callable[[Request], None | Awaitable[None]]

# Guard type
Guard = Callable[[ASGIConnection, BaseRouteHandler], bool | Awaitable[bool]]

# Cache types
CacheKeyBuilder = Callable[[Request], str]

# Type encoders
TypeEncodersMap = dict[Any, Callable[[Any], Any]]

Install with Tessl CLI