Endpoint Handlers ¶

Purpose: Handler Type & Behavior — What kind of operation each handler performs and how it executes.

Endpoint handlers are the functions that process requests and return responses. They're where your application logic lives—the "what to do" after routing determines "which handler to call."

The CQRS Pattern: Why Three Handler Types? ¶

Stario uses the Command Query Responsibility Segregation (CQRS) pattern to separate handlers by intent:

Query = "Give me information" (read-only, no side effects) → GET
Command = "Do something, I'll wait till you're done" (write operation, has side effects) → POST
Detached Command = "Do something, I'll check the status later" (async write, fire-and-forget) → POST with 204 No Content

Deep Dive: For comprehensive reasoning about CQRS, design philosophy, and real-world patterns, see The CQRS Pattern.

Endpoint Types: Query, Command, Detached Command ¶

Query Handlers (`@app.query()`) ¶

Purpose: Fetch and return data safely without side effects.

@app.query("/users")
async def list_users():
    """Retrieve all users"""
    return ...

@app.query("/users/{user_id}")
async def get_user(user_id: PathParam[int]):
    """Retrieve a specific user"""
    return ...

Characteristics:

HTTP method: GET (safe and idempotent)
Can be cached safely by browsers and servers
Multiple calls produce identical results
No state changes on the server
Client receives response immediately

Command Handlers (`@app.command()`) ¶

Purpose: Change state and return results to the client.

@app.command("/users")
async def create_user(name: QueryParam[str]):
    """Create a new user"""
    return ...

@app.command("/users/{user_id}")
async def update_user(user_id: PathParam[int], payload: JsonBody[UserUpdate]):
    """Update a user"""
    return ...

@app.command("/users/{user_id}")
async def delete_user(user_id: PathParam[int]):
    """Delete a user"""
    return None  # 204 No Content

Characteristics:

HTTP method: POST (can change state)
Not idempotent (repeated calls may have different effects)
Returns the result of the operation
Client waits for response (synchronous)
Cannot be safely cached

Detached Command Handlers (`@app.detached_command()`) ¶

Purpose: Fire-and-forget operations (start background work and return immediately).

@app.detached_command("/import")
async def import_data(file: UploadedFile):
    """Start background import - returns immediately"""
    await long_running_import(file)

@app.detached_command("/send-emails")
async def send_emails(user_ids: JsonBody[list[int]]):
    """Start background email sending"""
    await send_batch_emails(user_ids)

Characteristics:

HTTP method: POST
Returns 204 No Content immediately to the client
Handler continues running in the background
No response sent to the client
Useful for long-running operations

Two-Phase Execution ¶

Detached commands have a unique execution model that balances validation with responsiveness:

Phase 1: Dependencies in Foreground (Request-handling time)

All dependencies are resolved and validated during the request. If validation fails, the client gets an error response:

@app.detached_command("/process-file")
async def process_file(
    file: UploadedFile,              # ← Validated NOW (foreground)
    user_id: PathParam[int],         # ← Validated NOW
    db: Database,                    # ← Resolved NOW
):
    # If any dependency fails → exception raised → client gets error
    # Example: UploadedFile invalid → 422 response before handler starts

    # Handler body runs in background (Phase 2)
    await long_processing(file, user_id, db)

Phase 2: Handler Body in Background

Once all dependencies validate, Stario:

Returns 204 No Content to the client immediately
Continues executing the handler in the background
User's request is complete—they can proceed with other tasks

@app.detached_command("/send-notification")
async def send_notification(
    user_id: PathParam[int],         # Validated in foreground (Phase 1)
    db: Database,                    # Resolved in foreground
):
    """
    Timeline of execution:

    T1: Request arrives
    T2: user_id parsed and validated ✓
    T3: Database connection established ✓
    T4: All dependencies ready ✓
    T5: Return 204 No Content to client
    T6-T20: Handler body executes in background
            - Load user from database
            - Send notification
            - Update notification record
            - Client is long gone by now!
    """
    user = await db.get_user(user_id)  # Phase 2 (background)
    await send_email(user.email)
    await db.log_notification(user_id)

Key principle: Dependencies are your validation layer, the handler body is your execution layer.

Why this matters:

Validation happens immediately - User gets error response right away if command is malformed
Control returns to user quickly - If dependencies succeed, user gets 204 and continues
Processing happens safely - Actual work runs in background without blocking the request

Debugging Background Execution with Logging ¶

Since the handler runs in background, log what's happening to understand execution flow:

import logging

logger = logging.getLogger(__name__)

@app.detached_command("/import-data")
async def import_data(
    user_id: PathParam[int],
    db: Database,
    logger: Logger,  # Injected logger from Stario
):
    # Phase 1: Dependencies validated here (foreground)
    logger.info("Import requested", user_id=user_id)

    # Phase 2: Handler runs in background
    logger.info("Starting import process")

    items = await db.fetch_items()
    logger.info("Fetched items", count=len(items))

    for item in items:
        await process_item(item)
        logger.debug("Processed item", item_id=item.id)

    logger.info("Import completed successfully")

Logs include the request ID, allowing you to trace the entire background operation.

See Logging for comprehensive logging documentation.

Endpoint Handler Flow ¶

Stario's endpoint handling follows a clear three-phase flow:

1. Dependency Injection    → Resolve function arguments from the request
2. Handler Execution       → Your domain logic runs
3. Response Rendering      → Convert handler output to HTTP response

This separation lets you focus on business logic while the framework handles extracting request data and formatting responses.

Phase 1: Dependency Injection ¶

Before your handler runs, Stario inspects its parameters and resolves them from the request. This includes query parameters, path parameters, headers, request body, and custom dependencies.

@app.query("/users/{user_id}")
async def get_user(
    user_id: PathParam[int],         # ← Extracted from URL path
    include_posts: QueryParam[bool], # ← Extracted from query string
    db: Database,                    # ← Custom dependency injected
):
    # Dependencies are resolved and provided automatically
    user = await db.get_user(user_id)
    return render_user_profile(user, include_posts)

See Dependencies for comprehensive documentation on dependency injection.

Phase 2: Handler Execution ¶

Your function runs with all dependencies resolved. This is where your domain logic lives—business rules, data processing, orchestration.

@app.command("/posts")
async def create_post(
    title: Signal[str],
    content: Signal[str],
    user: CurrentUser,
    db: Database,
):
    # Your domain logic
    post = Post(title=title, content=content, author_id=user.id)
    await db.save(post)

    # Return what makes sense for your application
    return div(f"Post '{title}' created!")

Phase 3: Response Rendering ¶

Stario takes what you return or yield and converts it into an HTTP response. The conversion depends on what you return:

Response object → Sent directly (no rendering)
HTML content → Rendered to HTML string, wrapped in HTMLResponse
Generator (yields) → Converted to text/event-stream for Datastar
Falsey value (None, False)→ Returns 204 No Content

Handler Examples ¶

from stario import Stario
from stario.requests import QueryParam
from stario.html import h1, div, HtmlElement
from starlette.responses import JSONResponse

app = Stario()

@app.query("/")
async def html_response() -> HtmlElement:
    # A simple HTML response
    return h1("Welcome to Stario!")

@app.query("/search")
async def custom_response(page: QueryParam[int] = 1, limit: QueryParam[int] = 10) -> JSONResponse:
    # Return Response objects directly for full control
    return JSONResponse({"status": "success", "page": page, "limit": limit})

@app.command("/stream")
async def streaming_response():
    # Generators become streaming responses
    for i in range(10):
        yield "replace", "#counter", div(f"Counter: {i}")
        await asyncio.sleep(1)

Using route constructors instead (alternative):

from stario.routes import Query, Command

async def html_response():
    return h1("Welcome!")

async def custom_response(page: QueryParam[int]):
    return JSONResponse({"page": page})

app = Stario(
    Query("/", html_response),
    Query("/search", custom_response),
)

Response Rendering ¶

Depending on the return type, Stario will take different actions:

Returning `Response` Object Directly ¶

It's always possible to return a valid Response object directly. If you want to return a redirect response or custom 404—you can do it by returning a RedirectResponse or HTMLResponse with the appropriate status codes.

async def direct_response() -> JSONResponse:
    return JSONResponse({"status": "success"})

async def not_found_response() -> HTMLResponse:
    return HTMLResponse(content="<h1>404 Not Found</h1>", status_code=404)

When to Return Response Objects vs. Let Stario Render ¶

Return Response directly when you need:

Custom HTTP status codes (301 redirects, 410 Gone, 204 No Content)
Special headers (Content-Disposition for downloads, Cache-Control, Custom headers)
Non-HTML responses (JSON, XML, binary files, plain text)
Direct control over the entire response

Example - File Download with Custom Headers:

from starlette.responses import FileResponse

@app.query("/download/{file_id}")
async def download_file(file_id: PathParam[int]):
    # Return Response directly for special handling
    return FileResponse(
        path="/path/to/reports/file.pdf",
        media_type="application/pdf",
        headers={"Content-Disposition": "attachment; filename=report.pdf"}
    )

In all other cases, let Stario render HTML automatically:

Regular HTML pages → Return HTML elements
Empty responses → Return None for 204 No Content
Streaming responses → Use yield

This separation keeps your handlers focused on business logic while Stario handles response formatting.

Returning HTML Content ¶

When the function returns any other object than a Response or a generator, Stario will try to render it to HTML and return it as a valid HTML response.

If it's None, Stario will return a 204 No Content response. Otherwise, it will return a 200 OK response with the rendered HTML content.

Important: Only None returns a 204 No Content response. All other values—including False, 0, "", [], or {} - will be rendered as HTML text and returned as a 200 OK response.

async def html_response() -> HtmlElement:
    # A simple HTML response
    return h1("Welcome to Stario!")

async def none_response() -> None:
    # Returns 204 No Content ✅
    # Useful for commands that don't need to return anything
    return None

async def false_response() -> bool:
    # Returns 200 OK with "False" as content (not 204!) ⚠️
    return False

async def zero_response() -> int:
    # Returns 200 OK with "0" as content (not 204!) ⚠️
    return 0

Return Value	HTTP Status	Response Body	Notes
`None`	204 No Content	(empty)	✅ Only `None` returns 204
`False`	200 OK	`False`	Rendered as HTML text
`0`	200 OK	`0`	Rendered as HTML text
`""`	200 OK	(empty)	Rendered as empty HTML
`h1("Title")`	200 OK	`<h1>Title</h1>`	✅ Normal HTML element

Streaming Responses with `yield` ¶

When the function is a generator (uses yield), Stario will convert it to a streaming response compatible with Datastar SSE events.

The response will be a text/event-stream response with the appropriate headers.

Simplified format (recommended):

async def simplified_streaming():
    # Patch element (outer mode, fat-morph)
    yield div({"id": "counter"}, "Counter")

    # Patch signals
    yield {"counter": 42, "message": "Updated!"}

    # Execute script
    yield "script", "console.log('Hello from server!')"

    # Redirect
    yield "redirect", "/success"

    # Patch with specific mode and selector
    yield "replace", "#counter", div(42)

    # Remove element
    yield "remove", "#counter"

Object-based format (full Server-Sent Events control):

For comprehensive documentation on object-based streaming, Server-Sent Events (SSE) parameters, and all available yield types, see Response Interpretation.

Testing Your Handlers ¶

Use Starlette's TestClient to test your handlers without running a server:

from starlette.testclient import TestClient
from stario import Stario
from stario.requests import PathParam
from stario.html import h1, div

app = Stario()

@app.query("/")
async def home():
    return h1("Home")

@app.query("/users/{user_id}")
async def get_user(user_id: PathParam[int]):
    return div(f"User {user_id}")

client = TestClient(app)

# Test basic HTML response
response = client.get("/")
assert response.status_code == 200
assert "<h1>" in response.text

# Test with path parameters
response = client.get("/users/123")
assert response.status_code == 200
assert "User 123" in response.text

# Test with query parameters
response = client.get("/search?q=python&page=2")
assert response.status_code == 200
assert "Query: python" in response.text

Important: TestClient is synchronous, even for async handlers. Starlette handles the async/await conversion automatically.

For comprehensive testing documentation including patterns for dependencies, exceptions, headers, streaming, and more, see Testing.

Next Steps ¶

Learn about Dependencies for comprehensive dependency injection documentation
Understand The CQRS Pattern for deep design reasoning
Explore Testing for comprehensive testing strategies
Understand Datastar for streaming response details
Explore HTML for rendering with adapters