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

Plugin System

Name: tessl/pypi-litestar
Author: tessl

Extensible plugin architecture for integrating with external libraries and extending framework functionality. Litestar's plugin system includes core plugins and protocols for custom plugin development.

Capabilities

Plugin Protocols

Base protocols that define the plugin interface and lifecycle hooks.

class PluginProtocol(Protocol):
    """Base protocol for all plugins."""
    
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """
        Hook called during application initialization.

        Parameters:
        - app_config: Application configuration

        Returns:
        Modified application configuration
        """

class InitPluginProtocol(PluginProtocol):
    """Protocol for initialization plugins."""
    
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """Initialize plugin during app setup."""

class SerializationPluginProtocol(PluginProtocol):
    """Protocol for serialization plugins."""
    
    def supports_type(self, field_definition: FieldDefinition) -> bool:
        """
        Check if plugin supports serializing the given type.

        Parameters:
        - field_definition: Field definition to check

        Returns:
        True if plugin can handle the type
        """
    
    def create_dto_for_type(
        self,
        field_definition: FieldDefinition,
        handler_id: str,
        handler: BaseRouteHandler,
    ) -> type[AbstractDTO]:
        """
        Create DTO for the given type.

        Parameters:
        - field_definition: Field definition to create DTO for
        - handler_id: Unique handler identifier
        - handler: Route handler instance

        Returns:
        DTO class for the type
        """

class OpenAPISchemaPluginProtocol(PluginProtocol):
    """Protocol for OpenAPI schema generation plugins."""
    
    def is_plugin_supported_type(self, value: Any) -> bool:
        """Check if plugin supports the given type for schema generation."""
    
    def to_openapi_schema(
        self,
        field_definition: FieldDefinition,
        handler_id: str,
        handler: BaseRouteHandler,
    ) -> Schema:
        """
        Generate OpenAPI schema for the given type.

        Parameters:
        - field_definition: Field definition
        - handler_id: Handler identifier
        - handler: Route handler

        Returns:
        OpenAPI schema object
        """

class CLIPluginProtocol(PluginProtocol):
    """Protocol for CLI plugins."""
    
    def on_cli_init(self, cli: Group) -> None:
        """
        Hook called during CLI initialization.

        Parameters:
        - cli: Click CLI group to extend
        """

Core Plugin Classes

Built-in plugin implementations for common functionality.

class InitPlugin:
    def __init__(self, config: InitPluginConfig):
        """
        Initialization plugin for app setup.

        Parameters:
        - config: Plugin configuration
        """

    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """Initialize plugin during app setup."""

class SerializationPlugin:
    def __init__(self):
        """Base serialization plugin."""

    def supports_type(self, field_definition: FieldDefinition) -> bool:
        """Check if plugin supports the field type."""

    def create_dto_for_type(
        self,
        field_definition: FieldDefinition,
        handler_id: str,
        handler: BaseRouteHandler,
    ) -> type[AbstractDTO]:
        """Create DTO for the field type."""

class OpenAPISchemaPlugin:
    def __init__(self):
        """Base OpenAPI schema plugin."""

    def is_plugin_supported_type(self, value: Any) -> bool:
        """Check if plugin supports the type."""

    def to_openapi_schema(
        self,
        field_definition: FieldDefinition,
        handler_id: str,
        handler: BaseRouteHandler,
    ) -> Schema:
        """Generate OpenAPI schema."""

class CLIPlugin:
    def __init__(self, name: str):
        """
        CLI extension plugin.

        Parameters:
        - name: Plugin name
        """

    def on_cli_init(self, cli: Group) -> None:
        """Extend CLI with plugin commands."""

class DIPlugin:
    """Dependency injection plugin."""
    
    def has_typed_init(self, type_: type[Any]) -> bool:
        """Check if type has typed constructor."""
    
    def get_typed_init(self, type_: type[Any]) -> dict[str, Any]:
        """Get typed constructor signature."""

class ReceiveRoutePlugin:
    """Plugin for handling route reception."""
    
    def receive_route(self, route: HTTPRoute | WebSocketRoute) -> None:
        """Handle route registration."""

Plugin Registry

System for managing and coordinating multiple plugins.

class PluginRegistry:
    def __init__(self, plugins: Sequence[PluginProtocol] | None = None):
        """
        Plugin registry for managing plugins.

        Parameters:
        - plugins: Initial list of plugins
        """

    def add_plugin(self, plugin: PluginProtocol) -> None:
        """Add a plugin to the registry."""

    def remove_plugin(self, plugin: PluginProtocol) -> None:
        """Remove a plugin from the registry."""

    def get_plugins_of_type(self, plugin_type: type[T]) -> list[T]:
        """Get all plugins of a specific type."""

    def call_plugin_hooks(
        self,
        hook_name: str,
        *args: Any,
        **kwargs: Any,
    ) -> list[Any]:
        """Call a hook on all plugins that support it."""

Built-in Framework Plugins

Pre-built plugins for popular Python libraries and frameworks.

# Pydantic Plugin
class PydanticPlugin(SerializationPluginProtocol, OpenAPISchemaPluginProtocol):
    def __init__(self, prefer_alias: bool = False):
        """
        Pydantic integration plugin.

        Parameters:
        - prefer_alias: Use field aliases in serialization
        """

    def supports_type(self, field_definition: FieldDefinition) -> bool:
        """Check if type is a Pydantic model."""

    def create_dto_for_type(
        self,
        field_definition: FieldDefinition,
        handler_id: str,
        handler: BaseRouteHandler,
    ) -> type[PydanticDTO]:
        """Create PydanticDTO for model."""

# HTMX Plugin  
class HTMXPlugin(InitPluginProtocol):
    def __init__(self, config: HTMXConfig | None = None):
        """
        HTMX integration plugin.

        Parameters:
        - config: HTMX configuration
        """

    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """Initialize HTMX support."""

# Prometheus Plugin
class PrometheusPlugin(InitPluginProtocol):
    def __init__(self, config: PrometheusConfig | None = None):
        """
        Prometheus metrics plugin.

        Parameters:
        - config: Prometheus configuration
        """

    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """Initialize metrics collection."""

Plugin Configuration

Configuration classes for various plugins.

class InitPluginConfig:
    def __init__(
        self,
        *,
        connection_lifespan: Sequence[Callable[..., AsyncContextManager[None]]] | None = None,
        dto: type[AbstractDTO] | None = None,
        return_dto: type[AbstractDTO] | None = None,
        signature_namespace: dict[str, Any] | None = None,
        type_encoders: TypeEncodersMap | None = None,
    ):
        """
        Configuration for initialization plugins.

        Parameters:
        - connection_lifespan: Connection lifespan managers
        - dto: Default DTO for requests
        - return_dto: Default DTO for responses
        - signature_namespace: Signature inspection namespace
        - type_encoders: Type encoder mappings
        """

class HTMXConfig:
    def __init__(
        self,
        *,
        request_class: type[HTMXRequest] | None = None,
        response_class: type[HTMXTemplate] | None = None,
    ):
        """
        HTMX plugin configuration.

        Parameters:
        - request_class: Custom HTMX request class
        - response_class: Custom HTMX response class
        """

class PrometheusConfig:
    def __init__(
        self,
        *,
        app_name: str = "litestar",
        prefix: str = "litestar",
        labels: dict[str, str] | None = None,
        exclude_paths: set[str] | None = None,
        exclude_http_methods: set[str] | None = None,
        group_paths: bool = False,
        registry: CollectorRegistry | None = None,
    ):
        """
        Prometheus metrics configuration.

        Parameters:
        - app_name: Application name for metrics
        - prefix: Metric name prefix
        - labels: Default labels for metrics
        - exclude_paths: Paths to exclude from metrics
        - exclude_http_methods: HTTP methods to exclude
        - group_paths: Group similar paths together
        - registry: Custom metrics registry
        """

Usage Examples

Creating a Custom Plugin

from litestar.plugins import PluginProtocol
from litestar.config import AppConfig
import logging

class LoggingPlugin(PluginProtocol):
    """Plugin that configures request/response logging."""
    
    def __init__(self, logger_name: str = "litestar.requests"):
        self.logger_name = logger_name
        self.logger = logging.getLogger(logger_name)
    
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """Configure logging during app initialization."""
        
        # Add custom middleware for request logging
        def logging_middleware(app: ASGIApp) -> ASGIApp:
            async def middleware(scope: Scope, receive: Receive, send: Send) -> None:
                if scope["type"] == "http":
                    request = Request(scope, receive)
                    start_time = time.time()
                    
                    async def send_wrapper(message: Message) -> None:
                        if message["type"] == "http.response.start":
                            duration = time.time() - start_time
                            self.logger.info(
                                f"{request.method} {request.url.path} - "
                                f"{message['status']} ({duration:.3f}s)"
                            )
                        await send(message)
                    
                    await app(scope, receive, send_wrapper)
                else:
                    await app(scope, receive, send)
            return middleware
        
        # Add to middleware stack
        if app_config.middleware is None:
            app_config.middleware = []
        app_config.middleware.append(logging_middleware)
        
        return app_config

# Register the plugin
app = Litestar(
    route_handlers=[...],
    plugins=[LoggingPlugin("myapp.requests")]
)

Serialization Plugin

from litestar.plugins import SerializationPluginProtocol
from litestar.dto import AbstractDTO
import msgspec

class MsgspecPlugin(SerializationPluginProtocol):
    """Plugin for msgspec serialization support."""
    
    def supports_type(self, field_definition: FieldDefinition) -> bool:
        """Check if type is a msgspec Struct."""
        return (
            hasattr(field_definition.annotation, "__msgspec_struct__") or
            (hasattr(field_definition.annotation, "__origin__") and
             hasattr(field_definition.annotation.__origin__, "__msgspec_struct__"))
        )
    
    def create_dto_for_type(
        self,
        field_definition: FieldDefinition,
        handler_id: str,
        handler: BaseRouteHandler,
    ) -> type[AbstractDTO]:
        """Create MsgspecDTO for the struct type."""
        from litestar.dto import MsgspecDTO
        
        return MsgspecDTO[field_definition.annotation]

# Usage with msgspec models
@msgspec.defstruct
class Product:
    name: str
    price: float
    category: str

@post("/products")
def create_product(data: Product) -> Product:
    # Plugin automatically handles msgspec serialization
    return data

app = Litestar(
    route_handlers=[create_product],
    plugins=[MsgspecPlugin()]
)

OpenAPI Schema Plugin

from litestar.plugins import OpenAPISchemaPluginProtocol
from litestar.openapi.spec import Schema
from decimal import Decimal

class DecimalSchemaPlugin(OpenAPISchemaPluginProtocol):
    """Plugin for generating OpenAPI schema for Decimal types."""
    
    def is_plugin_supported_type(self, value: Any) -> bool:
        """Check if value is a Decimal type."""
        return value is Decimal or (
            hasattr(value, "__origin__") and value.__origin__ is Decimal
        )
    
    def to_openapi_schema(
        self,
        field_definition: FieldDefinition,
        handler_id: str,
        handler: BaseRouteHandler,
    ) -> Schema:
        """Generate schema for Decimal fields."""
        return Schema(
            type="string",
            format="decimal",
            pattern=r"^\d+(\.\d+)?$",
            example="99.99",
            description="Decimal number as string"
        )

# Usage
@dataclass
class Price:
    amount: Decimal
    currency: str

@post("/prices")
def create_price(data: Price) -> Price:
    return data

app = Litestar(
    route_handlers=[create_price],
    plugins=[DecimalSchemaPlugin()],
    openapi_config=OpenAPIConfig(title="Pricing API")
)

CLI Plugin

from litestar.plugins import CLIPluginProtocol
import click

class DatabaseCLIPlugin(CLIPluginProtocol):
    """Plugin that adds database management commands."""
    
    def __init__(self, db_url: str):
        self.db_url = db_url
    
    def on_cli_init(self, cli: click.Group) -> None:
        """Add database commands to CLI."""
        
        @cli.group()
        def db():
            """Database management commands."""
            pass
        
        @db.command()
        def migrate():
            """Run database migrations."""
            click.echo(f"Running migrations on {self.db_url}")
            # Migration logic here
        
        @db.command()
        def seed():
            """Seed database with initial data."""
            click.echo("Seeding database...")
            # Seeding logic here
        
        @db.command()
        @click.option("--confirm", is_flag=True, help="Confirm deletion")
        def reset(confirm: bool):
            """Reset database."""
            if not confirm:
                click.echo("Use --confirm to reset database")
                return
            click.echo("Resetting database...")
            # Reset logic here

# Register plugin
app = Litestar(
    route_handlers=[...],
    plugins=[DatabaseCLIPlugin("postgresql://localhost/myapp")]
)

# CLI commands available:
# litestar db migrate
# litestar db seed  
# litestar db reset --confirm

Using Built-in Plugins

from litestar.plugins.pydantic import PydanticPlugin
from litestar.plugins.prometheus import PrometheusPlugin, PrometheusConfig
from litestar.plugins.htmx import HTMXPlugin, HTMXConfig
from pydantic import BaseModel

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

# Pydantic plugin for automatic serialization
pydantic_plugin = PydanticPlugin(prefer_alias=True)

# Prometheus metrics plugin
prometheus_config = PrometheusConfig(
    app_name="myapp",
    prefix="myapp",
    labels={"service": "api", "version": "1.0"},
    exclude_paths={"/health", "/metrics"}
)
prometheus_plugin = PrometheusPlugin(prometheus_config)

# HTMX plugin for server-side rendering
htmx_plugin = HTMXPlugin()

@post("/users")
def create_user(data: User) -> User:
    # Pydantic plugin handles validation and serialization
    return data

@get("/metrics")
def metrics() -> str:
    # Prometheus plugin provides metrics endpoint
    from prometheus_client import generate_latest
    return generate_latest()

app = Litestar(
    route_handlers=[create_user, metrics],
    plugins=[pydantic_plugin, prometheus_plugin, htmx_plugin]
)

Plugin Registry Management

from litestar.plugins import PluginRegistry

# Create registry with initial plugins
registry = PluginRegistry([
    LoggingPlugin(),
    PydanticPlugin(),
    PrometheusPlugin()
])

# Add more plugins dynamically
registry.add_plugin(HTMXPlugin())

# Get plugins of specific types
serialization_plugins = registry.get_plugins_of_type(SerializationPluginProtocol)
cli_plugins = registry.get_plugins_of_type(CLIPluginProtocol)

# Call hooks on all plugins
results = registry.call_plugin_hooks(
    "on_app_init",
    app_config=AppConfig()
)

# Use registry with app
app = Litestar(
    route_handlers=[...],
    plugins=registry
)

Advanced Plugin with Dependency Injection

from litestar.plugins import InitPluginProtocol
from litestar.di import Provide

class DatabasePlugin(InitPluginProtocol):
    """Plugin that provides database connection to routes."""
    
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.pool = None
    
    async def create_pool(self):
        """Create database connection pool."""
        # Create connection pool (pseudo-code)
        self.pool = await create_pool(self.connection_string)
        return self.pool
    
    async def get_connection(self):
        """Get database connection from pool."""
        if not self.pool:
            await self.create_pool()
        return await self.pool.acquire()
    
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """Register database dependency."""
        
        # Add dependency provider
        if app_config.dependencies is None:
            app_config.dependencies = {}
        
        app_config.dependencies["db"] = Provide(self.get_connection)
        
        # Add lifespan handler for cleanup
        async def database_lifespan():
            try:
                yield
            finally:
                if self.pool:
                    await self.pool.close()
        
        if app_config.lifespan is None:
            app_config.lifespan = []
        app_config.lifespan.append(database_lifespan)
        
        return app_config

# Usage
@get("/users")
async def get_users(db=Dependency()) -> list[dict]:
    # Database connection injected by plugin
    result = await db.fetch("SELECT * FROM users")
    return [dict(row) for row in result]

app = Litestar(
    route_handlers=[get_users],
    plugins=[DatabasePlugin("postgresql://localhost/myapp")]
)

Types

# Plugin type union
PluginType = (
    PluginProtocol |
    InitPluginProtocol |
    SerializationPluginProtocol |
    OpenAPISchemaPluginProtocol |
    CLIPluginProtocol
)

# Configuration types
AppConfig = Any  # From litestar.config.app module
FieldDefinition = Any  # From litestar._signature module
BaseRouteHandler = Any  # From litestar.handlers module

# OpenAPI types
Schema = dict[str, Any]

# CLI types (from click)
Group = click.Group
Command = click.Command

# Prometheus types
CollectorRegistry = Any  # From prometheus_client

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

# Generic type variable for plugin types
T = TypeVar("T", bound=PluginProtocol)

Install with Tessl CLI