Dependencies ¶

Purpose: Data Injection - Automatically provide handlers with request data, services, and resources they need.

Dependencies provide a way to inject common functionality into your endpoint handlers - database connections, authentication state, configuration, logging, and more.

From a technical standpoint, dependencies are callables (sync or async functions) that Stario automatically resolves and provides to your handlers based on type annotations.

async def get_user(
    user_id: PathParam[int],      # ← Built-in dependency
    db: Database,                 # ← Custom dependency
    current_user: CurrentUser,    # ← Custom dependency
    logger: Logger,               # ← Built-in dependency
):
    logger.info("Fetching user", user_id=user_id, requester=current_user.id)
    user = await db.get_user(user_id)
    return render_user(user)

Dependency Graph ¶

The dependency graph is built once when a route is created, then reused for every request to that route.

Dependencies can nest recursively - a dependency can depend on other dependencies, which can depend on more dependencies, and so on. Stario resolves the entire graph automatically.

Why Use Annotated? ¶

Stario uses Python's typing.Annotated (similarly to FastAPI) to attach dependency information to type hints. This keeps your function signatures clean and readable:

# Clean signature: looks like normal Python
async def handler(user: CurrentUser, db: Database):
    pass

# Implementation detail: powered by Annotated
async def handler(
    user: Annotated[CurrentUser, get_current_user],
    db: Annotated[Database, get_database],
):
    pass

This design lets you:

• Keep signatures readable - Dependencies aren't cluttering the type hint
• Enable IDE support - Full autocomplete for dependency functions
• Support static analysis - Type checkers understand your code
• Make dependencies explicit - Clear where each value comes from

# Depends on nothing → independent
def get_settings() -> Settings:
    return load_settings_from_file()

# Depends on settings
def get_database(settings: Annotated[Settings, get_settings]) -> Database:
    return Database(settings.database_url)

# Depends on database
def get_user_repo(db: Annotated[Database, get_database]) -> UserRepository:
    return UserRepository(db)

# Depends on Request (special case)
def get_current_user_id(request: Request) -> int:
    return request.session.get("user_id")

# Depends on Stario app instance (special case)
def get_app_version(app: Stario) -> str:
    return app.state.version

# Can depend on any combination
async def handler(
    settings: Annotated[Settings, get_settings],
    db: Annotated[Database, get_database],
    user_repo: Annotated[UserRepository, get_user_repo],
    user_id: Annotated[int, get_current_user_id],
    version: Annotated[str, get_app_version],
):
    # All dependencies resolved and ready to use
    pass

Creating Custom Type Aliases ¶

For frequently-used dependencies, create custom Annotated type aliases to reduce boilerplate:

from typing import Annotated

# Define custom types once
type CurrentDB = Annotated[Database, get_database]
type CurrentUser = Annotated[User, get_current_user]
type AppSettings = Annotated[Settings, get_settings, "singleton"]

# Use them everywhere - much cleaner!
async def handler1(db: CurrentDB, user: CurrentUser):
    posts = await db.get_posts_by_user(user.id)
    return render_posts(posts)

async def handler2(settings: AppSettings, db: CurrentDB):
    await db.validate_against_settings(settings)
    return div("OK")

async def dashboard(user: CurrentUser, db: CurrentDB, settings: AppSettings):
    # Three dependencies, super clean signatures
    data = await db.get_dashboard_data(user.id, settings)
    return render_dashboard(user, data)

This pattern is especially useful for:

• Common dependencies - Used in many handlers
• Consistency - Same type alias everywhere means same dependency
• Readability - CurrentUser is clearer than Annotated[User, get_current_user]
• Refactoring - Change the dependency once, updates everywhere

Concurrent Resolution ¶

When possible, Stario resolves dependencies concurrently using asyncio.gather() or task groups. Independent async dependencies run in parallel, reducing total latency.

async def get_user_from_db(user_id: int) -> User:
    # Runs concurrently with get_posts_from_db if both are dependencies
    await asyncio.sleep(0.1)  # Simulate DB query
    return User(...)

async def get_posts_from_db(user_id: int) -> list[Post]:
    # Runs concurrently with get_user_from_db if both are dependencies
    await asyncio.sleep(0.1)  # Simulate DB query
    return [Post(...)]

async def handler(
    user: Annotated[User, get_user_from_db],
    posts: Annotated[list[Post], get_posts_from_db],
):
    # Both dependencies resolved in ~0.1s, not ~0.2s
    return render_profile(user, posts)

Concurrency Patterns ¶

Dependencies resolve in parallel by default. This is fast and usually correct. However, shared connections need care:

Problem: If your dependency connects to DB, and you resolve it twice, you share one connection:

# ❌ Problem: Both dependencies get the same connection
async def get_database() -> Database:
    return await connect_to_database()

async def handler(
    db1: Annotated[Database, get_database],  # Same connection
    db2: Annotated[Database, get_database],  # Same connection - concurrent issues!
):
    pass

# ✅ Solution: Use transient lifetime for isolated instances
async def handler(
    db1: Annotated[Database, get_database, "transient"],  # Different connection
    db2: Annotated[Database, get_database, "transient"],  # Different connection
):
    pass

In practice: Most dependencies are immutable (configs, services, repositories). Connection pools are the exception. Don't over-optimize for this edge case.

Dependency Order: Dependencies resolve in parallel, not in the order you define them. If you need sequential setup or shared state, nest dependencies or use a single dependency that manages ordering internally.

Dependency Lifetime ¶

Dependencies can have different lifetimes that control when they're called and how they're cached:

"request" (Default) ¶

Called once and cached for the duration of a single request. Reused across all dependencies and subdependencies within the same request.

async def get_database() -> Database:
    # Called once per request, then reused
    return await connect_to_database()

async def handler(
    db1: Annotated[Database, get_database],
    db2: Annotated[Database, get_database],
):
    # db1 and db2 are the same instance
    assert db1 is db2

Use for: Database connections, current user, request-scoped caches, anything that should be shared within a request but not across requests.

"transient" ¶

Called every time it's used-no caching. Can be called multiple times in the same request.

async def get_database_transient() -> Database:
    # Called every time it's needed
    return await connect_to_database()

async def handler(
    db1: Annotated[Database, get_database_transient, "transient"],
    db2: Annotated[Database, get_database_transient, "transient"],
):
    # db1 and db2 are different instances
    assert db1 is not db2

Use for: When you need multiple independent instances, avoiding concurrent access issues, or when caching would cause problems.

"singleton" ¶

Called once and cached for the entire lifetime of the application. Reused across all requests.

def get_settings() -> Settings:
    # Called once when the app starts, then reused forever
    return load_settings_from_file()

async def handler(
    settings: Annotated[Settings, get_settings, "singleton"],
):
    # Same settings instance for every request
    pass

Use for: Configuration, connection pools, read-only data, anything that's expensive to create and safe to share across requests.

Note: Avoid global variables by using singleton dependencies instead-it makes testing easier and dependencies explicit.

"lazy" ¶

Returns an awaitable that you can await to get the actual dependency instance. Defers execution until you explicitly await it.

async def expensive_api_call() -> Data:
    # Only called when you await the lazy dependency
    return await fetch_from_external_api()

async def handler(
    data_promise: Annotated[Awaitable[Data], expensive_api_call, "lazy"],
):
    # At this point, expensive_api_call has NOT been called yet

    # Check some condition first
    if should_fetch:
        data = await data_promise  # ← Only now does it run
        return render_with_data(data)
    else:
        return render_without_data()

Real-world example: Background task execution

import asyncio
from typing import Annotated, Awaitable

async def expensive_dependency(logger: Logger):
    """A slow dependency that should run in background"""
    logger.info("Dependency called")
    await asyncio.sleep(1.5)
    logger.info("Dependency resolved after delay")
    return "dependency_result"

@app.detached_command("/action")
async def background_action(
    logger: Logger,
    dependency: Annotated[Awaitable[str], expensive_dependency, "lazy"]
):
    # Dependencies are validated, user gets 204 immediately
    logger.info("Action called")

    # Handler runs in background - await dependency when needed
    dep_value = await dependency
    logger.info("Dependency resolved", result=dep_value)

    await asyncio.sleep(10)  # Long-running operation
    logger.info("Background action complete")

In this pattern:

1. User gets immediate 204 No Content response
2. Handler runs as background task
3. Lazy dependency waits until explicitly awaited
4. Slow operations don't block the client

If you have dependencies with subdependencies, they're all resolved when you await:

def get_database() -> Database:
    return Database()

def get_repository(db: Annotated[Database, get_database]) -> UserRepository:
    return UserRepository(db)

async def handler(
    repo_promise: Annotated[Awaitable[UserRepository], get_repository, "lazy"],
):
    # Neither get_database nor get_repository has been called

    if need_users:
        repo = await repo_promise  # ← Both get_database and get_repository run now
        users = await repo.get_all()
        return render_users(users)
    else:
        return render_empty()

Use for: Expensive operations that might not be needed, conditional data fetching, deferring work until after checks.

Special Cases ¶

• Request is always "request" lifetime
• Stario app instance is always "singleton" lifetime

Built-in Dependencies ¶

Stario provides built-in dependencies in the stario.requests module for common request data extraction. They're type aliases for Annotated declarations with extractors (or parsers for Pydantic validation).

Query Parameters ¶

from stario.requests import QueryParam, QueryParams

async def search(
    q: QueryParam[str],           # Reads 'q' from query string
    page: QueryParam[int],         # Parses 'page' to int
    tags: QueryParams[str],        # Reads multiple 'tag' parameters
):
    # q = "python" from ?q=python
    # page = 2 from ?page=2
    # tags = ["web", "api"] from ?tag=web&tag=api
    pass

Named parameters:

search_term: Annotated[str, ParseQueryParam(name="query")]
# Reads 'query' param but names it 'search_term'

See also: Path Parameters, Headers

Path Parameters ¶

from stario.requests import PathParam

# Route: /users/{user_id}
async def get_user(user_id: PathParam[int]):
    # user_id = 123 from /users/123
    pass

See also: Query Parameters, Endpoints

Headers ¶

from stario.requests import Header, Headers

async def handler(
    auth: Header[str],              # Reads 'Authorization' header
    accept: Headers[list[str]],     # Reads 'Accept' header, splits on comma
):
    # auth = "Bearer token123"
    # accept = ["text/html", "application/json"]
    pass

See also: Cookies, Request Body, Dependencies

Cookies ¶

from stario.requests import Cookie

async def handler(session_id: Cookie[str]):
    # session_id = "abc123" from cookie
    pass

See also: Headers, Request Body

Request Body ¶

from stario.requests import Body, JsonBody, RawBody
from pydantic import BaseModel

class CreateUserPayload(BaseModel):
    username: str
    email: str

async def create_user(payload: JsonBody[CreateUserPayload]):
    # Automatically parsed and validated by Pydantic
    user = User(username=payload.username, email=payload.email)
    await save_user(user)

Body types:

• Body[bytes] - Raw bytes
• RawBody[str] - Raw string
• JsonBody[T] - Parsed JSON, validated with Pydantic's TypeAdapter

Note: Parsing uses Pydantic's TypeAdapter, expecting values in valid JSON format. Behavior might differ from intuition for complex types.

See also: Pydantic Validation, Endpoints

Datastar Integration ¶

Datastar has its own comprehensive reference documentation. See Datastar for:

• Reading Signal Values - How to extract signals from requests (Signal, Signals)
• Signal Format - How signals are sent in GET vs POST requests
• Testing Signals - How to test handlers that use signal dependencies
• Attributes Builder - Using the Datastar dependency to build interactive elements

For a quick overview of how signals work as dependencies, the key idea is: - Signal[T] reads a single signal value, parsed to type T - Signals reads all signals as a dictionary - Nested signals use double underscores: user__name for user.name

Example:

from stario.datastar import Signal, Signals

async def handler(counter: Signal[int], all_signals: Signals):
    # counter: single signal parsed as int
    # all_signals: dictionary of all signals
    return div(f"Counter: {counter}, Total: {len(all_signals)}")

For complete details including format specifications, nested signals, and testing patterns, see Datastar Reference.

Custom Dependencies ¶

Create dependencies by writing functions that return what your handlers need:

from stario.requests import Request
from typing import Annotated

async def get_database(request: Request) -> Database:
    # Access app state for connection pool
    pool = request.app.state.db_pool
    conn = await pool.acquire()
    return conn  # Return connection

async def get_current_user(
    request: Request,
    db: Annotated[Database, get_database],
) -> User:
    user_id = request.session.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401)
    return await db.get_user(user_id)

# Use in handlers
async def dashboard(
    user: Annotated[User, get_current_user],
    db: Annotated[Database, get_database],
):
    posts = await db.get_user_posts(user.id)
    return render_dashboard(user, posts)

Dynamically Resolved Dependencies ¶

When a dependency function expects a single parameter with type annotation inspect.Parameter, Stario treats it as a dynamically resolved dependency. This means the dependency builder receives metadata about the parameter where it's being used - including the parameter name, type annotation, and default value.

This pattern is used by built-in dependencies like QueryParam, PathParam, and Header to automatically detect the parameter name they should extract from the request:

from inspect import Parameter as InspectParameter

class ParseQueryParam:
    """A dynamically resolved dependency"""
    def __init__(self, name: str | None = None):
        self.name = name

    def __call__(self, param: InspectParameter) -> Callable:
        # param contains metadata about where this dependency is used:
        # - param.name: The parameter name ("search_term", "page", etc.)
        # - param.annotation: The type annotation (str, int, etc.)
        # - param.default: The default value if provided

        param_name = self.name or param.name  # Use custom name or infer from parameter

        def extractor(request: Request) -> Any:
            # Now we know exactly what to extract and what type to validate
            value = request.query_params.get(param_name)
            # ... validate and return ...

        return extractor

This allows you to write clean handler signatures where the parameter name itself conveys meaning:

# ✅ Clean: Parameter name is self-documenting
async def search(
    search_term: QueryParam[str],    # Extracts ?search_term=...
    page: QueryParam[int] = 1,       # Extracts ?page=..., defaults to 1
    sort_by: QueryParam[str] = "relevance",
):
    return render_results(search_term, page, sort_by)

# This is equivalent to manually specifying the names:
async def search(
    search_term: Annotated[str, ParseQueryParam(name="search_term")],
    page: Annotated[int, ParseQueryParam(name="page")] = 1,
    sort_by: Annotated[str, ParseQueryParam(name="sort_by")] = "relevance",
):
    return render_results(search_term, page, sort_by)

When building custom dynamically resolved dependencies:

from inspect import Parameter as InspectParameter
from typing import Annotated, Callable, Any

class CustomExtractor:
    """Custom dynamically resolved dependency that adapts to its usage context"""

    def __init__(self, prefix: str = ""):
        self.prefix = prefix

    def __call__(self, param: InspectParameter) -> Callable:
        # param.name is the parameter name where this is used
        # param.annotation is the type hint
        # param.default is the default value

        lookup_key = f"{self.prefix}{param.name}" if self.prefix else param.name

        def extractor(request: Request) -> Any:
            # Use lookup_key to extract the right value
            return request.state.get(lookup_key)

        return extractor

# Usage: Parameter name becomes part of the extraction logic
async def handler(
    user_id: Annotated[int, CustomExtractor()],           # Looks up "user_id"
    org_id: Annotated[int, CustomExtractor(prefix="org_")],  # Looks up "org_org_id"
):
    pass

Dependencies with Cleanup (Context Managers) ¶

Dependencies can support the context manager protocol (__enter__/__exit__ or __aenter__/__aexit__) for automatic cleanup. When a dependency returns a context manager instance, Stario automatically manages its lifecycle.

Asynchronous Context Manager:

from contextlib import asynccontextmanager

class DatabaseConnection:
    async def __aenter__(self):
        print("Opening database connection")
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing database connection")
        await self.close()

async def get_database() -> DatabaseConnection:
    """Returns a context manager instance"""
    db = DatabaseConnection()
    return db  # Stario will handle __aenter__/__aexit__

@app.query("/data")
async def fetch_data(db: Annotated[DatabaseConnection, get_database]):
    # Connection is automatically opened via __aenter__
    data = await db.query("SELECT * FROM users")
    # Connection is automatically closed via __aexit__ after handler completes
    return render_data(data)

Synchronous Context Manager:

from contextlib import contextmanager

class FileHandle:
    def __enter__(self):
        print("Opening file")
        self.file = open("data.txt", "r")
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Closing file")
        self.file.close()

@app.query("/file")
async def read_file(file: Annotated[FileHandle, FileHandle]):
    # File is automatically opened via __enter__
    content = file.read()
    # File is automatically closed via __exit__ after handler completes
    return render_content(content)

Key Points:

• The dependency function returns a context manager instance (not yields)
• Stario automatically calls __aenter__ (or __enter__) before providing the dependency
• Stario automatically calls __aexit__ (or __exit__) after the request completes
• Works with any number of context manager dependencies in the same handler
• Cleanup happens even if exceptions occur during request handling

Using contextlib Helpers:

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_database():
    """Using contextlib decorator - returns async context manager"""
    db = await create_connection()
    try:
        yield db  # Provide the dependency
    finally:
        await db.close()

@app.query("/users")
async def list_users(db: Annotated[Database, get_database]):
    # Database connection is automatically managed
    users = await db.get_users()
    return render_users(users)

See Also ¶

• Datastar Reference - Comprehensive guide to signals and reactive features
• Endpoints - How dependencies fit into handler execution
• Logging - Logger dependency injection
• Dependency Injection in Action - Comparison of the example implementation using dependency injection vs manual implementation