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

Request and Response

Name: tessl/pypi-litestar
Author: tessl

Connection objects for accessing request data and generating responses. The Request object provides access to HTTP request data including headers, body, query parameters, and path parameters. Response objects handle returning different content types and status codes.

Capabilities

Request Object

The Request object provides access to all aspects of an incoming HTTP request through a rich API for headers, body parsing, parameters, and more.

class Request(ASGIConnection):
    def __init__(self, scope: Scope, receive: Receive = empty_receive):
        """
        Initialize a Request object.

        Parameters:
        - scope: ASGI scope dictionary
        - receive: ASGI receive callable
        """

    # Body parsing methods
    async def body(self) -> bytes:
        """Get the raw request body as bytes."""

    async def json(self) -> Any:
        """Parse request body as JSON."""

    async def form(self) -> dict[str, str | list[str]]:
        """Parse request body as form data."""

    async def msgpack(self) -> Any:
        """Parse request body as MessagePack."""

    async def stream(self) -> AsyncIterator[bytes]:
        """Stream the request body in chunks."""

    # Parameter access
    @property
    def query_params(self) -> MultiDict[str, str]:
        """Get query parameters as a MultiDict."""

    @property
    def path_params(self) -> dict[str, Any]:
        """Get path parameters with type conversion."""

    # Headers and cookies
    @property
    def headers(self) -> Headers:
        """Get request headers."""

    @property
    def cookies(self) -> dict[str, str]:
        """Get request cookies."""

    # Request metadata
    @property
    def method(self) -> Method:
        """Get HTTP method."""

    @property
    def url(self) -> URL:
        """Get request URL."""

    @property
    def content_type(self) -> tuple[str, dict[str, str]]:
        """Get content type and parameters."""

    @property
    def client(self) -> Address | None:
        """Get client address."""

    @property
    def auth(self) -> Any:
        """Get authentication information."""

    @property
    def user(self) -> Any:
        """Get authenticated user."""

    @property
    def state(self) -> State:
        """Get request state."""

    @property
    def session(self) -> dict[str, Any]:
        """Get session data."""

    # Request matching
    def url_for(self, name: str, **path_parameters: Any) -> str:
        """Generate URL for named route."""

    def url_for_static_asset(self, name: str, file_path: str) -> str:
        """Generate URL for static asset."""

Response Objects

Response objects handle different content types and response configurations.

class Response:
    def __init__(
        self,
        content: Any = None,
        *,
        status_code: int = 200,
        headers: ResponseHeaders | None = None,
        media_type: MediaType | str | None = None,
        background: BackgroundTask | BackgroundTasks | None = None,
        cookies: Sequence[Cookie] | None = None,
        encoding: str = "utf-8",
    ):
        """
        Create a Response object.

        Parameters:
        - content: Response content (automatically serialized)
        - status_code: HTTP status code
        - headers: Response headers
        - media_type: Content media type
        - background: Background task(s) to execute after response
        - cookies: Response cookies
        - encoding: Content encoding
        """

    @property
    def content(self) -> bytes:
        """Get response content as bytes."""

    def set_header(self, key: str, value: str) -> None:
        """Set a response header."""

    def set_cookie(
        self,
        key: str,
        value: str | None = None,
        *,
        max_age: int | None = None,
        expires: int | datetime | None = None,
        path: str = "/",
        domain: str | None = None,
        secure: bool = False,
        httponly: bool = False,
        samesite: Literal["lax", "strict", "none"] = "lax",
    ) -> None:
        """Set a response cookie."""

    def delete_cookie(
        self,
        key: str,
        path: str = "/",
        domain: str | None = None,
    ) -> None:
        """Delete a cookie."""

File Response

Response for serving files with proper headers and streaming support.

class File(Response):
    def __init__(
        self,
        path: str | Path,
        *,
        filename: str | None = None,
        stat_result: os.stat_result | None = None,
        chunk_size: int = 1024 * 1024,
        content_disposition_type: Literal["attachment", "inline"] = "attachment",
        etag: ETag | None = None,
        **kwargs: Any,
    ):
        """
        Create a file response.

        Parameters:
        - path: Path to the file
        - filename: Override filename in Content-Disposition header
        - stat_result: Pre-computed file stat result
        - chunk_size: Size of chunks for streaming
        - content_disposition_type: Content disposition type
        - etag: ETag configuration
        """

Streaming Response

Response for streaming data to the client.

class Stream(Response):
    def __init__(
        self,
        iterator: Iterator[str | bytes] | AsyncIterator[str | bytes],
        *,
        media_type: MediaType | str = MediaType.TEXT,
        **kwargs: Any,
    ):
        """
        Create a streaming response.

        Parameters:
        - iterator: Iterator yielding response chunks
        - media_type: Content media type
        """

Redirect Response

Response for HTTP redirects.

class Redirect(Response):
    def __init__(
        self,
        path: str,
        *,
        status_code: int = HTTP_302_FOUND,
        **kwargs: Any,
    ):
        """
        Create a redirect response.

        Parameters:
        - path: Redirect target URL
        - status_code: HTTP redirect status code
        """

Template Response

Response for rendered templates.

class Template(Response):
    def __init__(
        self,
        name: str,
        context: dict[str, Any] | None = None,
        *,
        **kwargs: Any,
    ):
        """
        Create a template response.

        Parameters:
        - name: Template name
        - context: Template context variables
        """

Server-Sent Events

Response for server-sent events streaming.

class ServerSentEvent(Response):
    def __init__(
        self,
        iterator: Iterator[ServerSentEventMessage] | AsyncIterator[ServerSentEventMessage],
        *,
        **kwargs: Any,
    ):
        """
        Create a server-sent events response.

        Parameters:
        - iterator: Iterator yielding SSE messages
        """

class ServerSentEventMessage:
    def __init__(
        self,
        data: str | bytes | dict | list,
        *,
        event: str | None = None,
        id: str | None = None,
        retry: int | None = None,
        comment: str | None = None,
    ):
        """
        Create an SSE message.

        Parameters:
        - data: Message data
        - event: Event type
        - id: Message ID
        - retry: Retry interval in milliseconds
        - comment: Comment text
        """

Data Structures

Core data structures used in requests and responses.

class Headers(CaseInsensitiveDict[str]):
    """Case-insensitive HTTP headers dictionary."""
    
    def __init__(self, headers: Mapping[str, str] | RawHeaders | None = None):
        """Initialize headers from mapping or raw headers."""

    def add(self, key: str, value: str) -> None:
        """Add a header value (allows duplicates)."""

    def get_list(self, key: str) -> list[str]:
        """Get all values for a header as a list."""

class MultiDict(dict[str, str]):
    """Dictionary supporting multiple values per key."""
    
    def getlist(self, key: str) -> list[str]:
        """Get all values for a key as a list."""

    def add(self, key: str, value: str) -> None:
        """Add a value for a key."""

class Cookie:
    def __init__(
        self,
        key: str,
        value: str | None = None,
        *,
        max_age: int | None = None,
        expires: int | datetime | None = None,
        path: str = "/",
        domain: str | None = None,
        secure: bool = False,
        httponly: bool = False,
        samesite: Literal["lax", "strict", "none"] = "lax",
    ):
        """Create a cookie."""

class URL:
    def __init__(self, url: str = ""):
        """Parse a URL string."""

    @property
    def scheme(self) -> str:
        """Get URL scheme."""

    @property
    def hostname(self) -> str | None:
        """Get hostname."""

    @property
    def port(self) -> int | None:
        """Get port number."""

    @property
    def path(self) -> str:
        """Get URL path."""

    @property
    def query(self) -> str:
        """Get query string."""

    @property
    def fragment(self) -> str:
        """Get URL fragment."""

    def replace(self, **kwargs: Any) -> URL:
        """Create a new URL with replaced components."""

Usage Examples

Request Data Access

from litestar import get, post, Request
from litestar.exceptions import ValidationException

@get("/info")
async def request_info(request: Request) -> dict:
    return {
        "method": request.method,
        "url": str(request.url),
        "headers": dict(request.headers),
        "query_params": dict(request.query_params),
        "client": request.client.host if request.client else None,
    }

@get("/search")
async def search(request: Request) -> dict:
    query = request.query_params.get("q", "")
    limit = int(request.query_params.get("limit", "10"))
    
    return {
        "query": query,
        "limit": limit,
        "results": []  # Search results would go here
    }

@post("/data")
async def handle_data(request: Request) -> dict:
    content_type = request.headers.get("content-type", "")
    
    if "application/json" in content_type:
        data = await request.json()
    elif "application/x-www-form-urlencoded" in content_type:
        data = await request.form()
    else:
        raise ValidationException("Unsupported content type")
    
    return {"received": data}

Custom Response Types

from litestar import get
from litestar.response import Response, File, Stream, Redirect
from litestar.status_codes import HTTP_201_CREATED
import json

@get("/custom")
async def custom_response() -> Response:
    data = {"message": "Custom response"}
    return Response(
        content=json.dumps(data),
        status_code=HTTP_201_CREATED,
        headers={"X-Custom-Header": "value"},
        media_type="application/json"
    )

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

@get("/stream-data")
async def stream_data() -> Stream:
    async def generate_data():
        for i in range(100):
            yield f"chunk {i}\n"
    
    return Stream(
        iterator=generate_data(),
        media_type="text/plain"
    )

@get("/redirect-to-docs")
async def redirect_to_docs() -> Redirect:
    return Redirect("https://docs.litestar.dev")

Cookie Handling

from litestar import get, post, Request, Response

@post("/login")
async def login(request: Request) -> Response:
    # Authenticate user (simplified)
    user_data = await request.json()
    
    response = Response({"status": "logged in"})
    response.set_cookie(
        "session_id",
        "abc123",
        max_age=3600,  # 1 hour
        httponly=True,
        secure=True,
        samesite="strict"
    )
    return response

@get("/profile")
async def get_profile(request: Request) -> dict:
    session_id = request.cookies.get("session_id")
    if not session_id:
        return {"error": "Not authenticated"}
    
    return {"user": "alice", "session": session_id}

@post("/logout")
async def logout() -> Response:
    response = Response({"status": "logged out"})
    response.delete_cookie("session_id")
    return response

Server-Sent Events

from litestar import get
from litestar.response import ServerSentEvent, ServerSentEventMessage
import asyncio

@get("/events")
async def event_stream() -> ServerSentEvent:
    async def generate_events():
        counter = 0
        while True:
            yield ServerSentEventMessage(
                data={"timestamp": time.time(), "counter": counter},
                event="update",
                id=str(counter)
            )
            counter += 1
            await asyncio.sleep(1)
    
    return ServerSentEvent(generate_events())

File Upload Handling

from litestar import post, Request
from litestar.datastructures import UploadFile

@post("/upload")
async def upload_file(request: Request) -> dict:
    form_data = await request.form()
    
    uploaded_file = form_data.get("file")
    if isinstance(uploaded_file, UploadFile):
        content = await uploaded_file.read()
        return {
            "filename": uploaded_file.filename,
            "size": len(content),
            "content_type": uploaded_file.content_type
        }
    
    return {"error": "No file uploaded"}

Types

# ASGI types
Scope = dict[str, Any]
Receive = Callable[[], Awaitable[Message]]
Send = Callable[[Message], Awaitable[None]]
Message = dict[str, Any]

# HTTP types
Method = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS", "TRACE"]
ResponseHeaders = Mapping[str, str] | Sequence[tuple[str, str]]
RawHeaders = list[tuple[bytes, bytes]]

# Content types
ContentType = tuple[str, dict[str, str]]

# Iterator types for streaming
SyncIterator = Iterator[str | bytes]
AsyncIterator = AsyncIterator[str | bytes]

Install with Tessl CLI