Exceptions ¶

Exceptions are how you signal errors in your application. Stario provides a structured way to handle them, ensuring clients receive appropriate responses and errors are properly logged.

Exception Handling Architecture ¶

Stario uses a two-layer exception handling system:

ExceptionMiddleware (inner layer)

• Catches exceptions raised in your handlers
• Routes to registered exception handlers (if any exist)
• Only handles exceptions it has registered handlers for

GuardianMiddleware (outer layer)

• Catches ALL unhandled exceptions that escape ExceptionMiddleware
• Generates request IDs for tracing
• Logs all exceptions with full context
• Ensures clients receive proper 500 responses instead of broken connections
• Handles exceptions after response has started (streaming errors)

Why two layers?

The separation allows:

1. ExceptionMiddleware to handle known, recoverable errors efficiently (quick path)
2. GuardianMiddleware to catch unexpected errors and prevent server crashes (safety net)
3. Request IDs to be generated at the outermost layer, making them available throughout the entire request lifecycle
4. Comprehensive logging with request context for debugging

Exception Flow Through Middleware ¶

Handler raises exception
    ↓
[ExceptionMiddleware] ← Tries registered handlers first
    ↓ (if not handled)
[Your Middlewares] ← Pass through
    ↓
[GuardianMiddleware] ← Catches ALL unhandled exceptions, logs with request ID, returns 500

What Happens When You Raise an Exception ¶

In Regular Handlers ¶

When you raise an exception before returning, Stario's exception handling catches it:

from starlette.exceptions import HTTPException

async def get_user(user_id: PathParam[int], db: Database):
    user = await db.get_user(user_id)

    if not user:
        # Raise an HTTP exception
        raise HTTPException(status_code=404, detail="User not found")

    return render_user(user)

What happens:

1. Exception is raised
2. Caught by ExceptionMiddleware (if handler registered) or GuardianMiddleware
3. Routed to registered exception handler (if any)
4. Response sent to client
5. Exception logged with request ID by GuardianMiddleware

In Generators (Streaming / Response Already Started) ¶

This is where things get dangerous. When your handler is a generator (uses yield), the response starts as soon as the first event is yielded. After that, you cannot change the status code or headers—they've already been sent to the client.

Why? HTTP/SSE specification requires status codes and headers to be sent at the start of the response. Once data starts streaming, it's too late to change them.

async def streaming_handler():
    yield "replace", "#status", div("Loading...")

    # ⚠️ Response has started! Status code is 200, headers sent.
    # HTTP spec: Headers sent at stream start, cannot be changed.

    try:
        data = await fetch_data()  # Might fail
        yield "replace", "#content", render_data(data)
    except Exception as exc:
        # ❌ Too late to return a 500 status code!
        # The client already received 200 OK
        # Best you can do: send error content in the stream
        yield "replace", "#content", div(
            {"class": "error"},
            f"Error: {exc}"
        )

Key Points:

• Once you yield, the HTTP response starts (status 200, headers sent)
• Raising exceptions after this point can't change the status code
• The exception is still logged by GuardianMiddleware, but the client sees 200 OK with error content
• GuardianMiddleware marks this with response_started=True in logs for visibility

Best practices for streaming handlers:

async def safe_streaming_handler(query: QueryParam[str]):
    try:
        # Step 1: Validate EVERYTHING BEFORE first yield
        if not query:
            raise HTTPException(status_code=400, detail="Query required")

        # Fetch and validate all required data
        data = await fetch_from_database(query)
        if not data:
            raise HTTPException(status_code=404, detail="No results found")

    except HTTPException:
        # ✅ Still time to raise - response hasn't started yet
        raise

    # Step 2: NOW we can start streaming - everything is validated
    yield div({"id": "status"}, f"Found {len(data)} results...")

    # Step 3: Stream the data - handle per-item errors gracefully
    for item in data:
        try:
            result = process_item(item)
            yield div({"class": "result"}, str(result))
        except ProcessingError as e:
            # Individual items failing is OK - stream the error
            yield div({"class": "error"}, f"Failed: {item}: {e}")

Safe Streaming Patterns ¶

The key to safe streaming handlers is validating all required data BEFORE the first yield. Once you yield, the response status code is locked in at 200 OK (this is HTTP/SSE spec, not a Stario limitation).

See the detailed example in the section above ("In Generators") for the recommended validate-before-yield pattern. The key principle: all validation must happen before the first yield, then handle per-item errors gracefully in the stream.

Best Practices for Validation in Streaming ¶

Before the first yield, validate:

• All required parameters are present and valid type
• User is authenticated (if needed)
• User has required permissions
• Database connections can be established
• External API endpoints are reachable
• Required files exist and are readable
• All business logic validation passes

By validating everything upfront, you ensure that once streaming starts, you only need to handle per-item errors or unexpected edge cases—not fundamental failures.

After the first yield, remember:

• Only stream errors in the response, never raise HTTPException
• Use yield div({"class": "error"}, ...) to show errors in the stream
• HTTP status is already 200 and locked in—you cannot change it
• All exceptions should yield error content, not propagate

Default Exception Behavior ¶

Without custom handlers, Stario handles exceptions as follows:

HTTPException (from Starlette) ¶

from starlette.exceptions import HTTPException

raise HTTPException(status_code=404, detail="Not found")
# → 404 response with "Not found" in body

Unhandled Exceptions ¶

raise ValueError("Something went wrong")
# → Caught by GuardianMiddleware
# → 500 Internal Server Error response
# → Exception logged with full traceback and request ID

Exception Types and Where They're Caught ¶

Different exceptions are caught at different layers of Stario, with different outcomes:

Important: You Cannot Catch Parameter Validation Errors ¶

# ❌ This WILL NOT catch validation errors (422)
async def handler(page: QueryParam[int]):
    try:
        # This try/except doesn't catch validation errors!
        # If ?page=abc, validation happens BEFORE this code runs
        pass
    except ValueError:
        # Won't execute for ?page=abc
        pass

Why? Pydantic validation happens during dependency resolution, which occurs before your handler function is called. Your try/except block only catches exceptions from code you write inside the handler.

When You CAN Use Try/Except ¶

# ✅ This WILL catch errors from your own code
async def handler(page: QueryParam[int]):
    try:
        # page is already validated (int type)
        data = await fetch_page(page)  # YOUR code here
    except ValueError:
        # Catches errors from fetch_page, not from parameter parsing
        raise HTTPException(status_code=400, detail="Invalid page data")

Custom Exception Handlers ¶

Register custom handlers to control how specific exceptions are handled:

from stario import Stario
from stario.html import div, h1, p, a
from stario.html.core import render
from starlette.responses import HTMLResponse
from starlette.requests import Request

async def not_found_handler(request: Request, exc: HTTPException):
    """Custom 404 page"""
    return HTMLResponse(
        content=render(
            div(
                h1("Page Not Found"),
                p(f"The page '{request.url.path}' doesn't exist."),
                a({"href": "/"}, "Go home"),
            )
        ),
        status_code=404,
    )

async def value_error_handler(request: Request, exc: ValueError):
    """Handle validation errors"""
    return HTMLResponse(
        content=render(
            div(
                h1("Invalid Input"),
                p(str(exc)),
            )
        ),
        status_code=400,
    )

app = Stario(
    exception_handlers={
        404: not_found_handler,
        ValueError: value_error_handler,
    }
)

Handler Signature ¶

Exception handlers must match this signature:

async def handler(request: Request, exc: Exception) -> Response:
    # request: The incoming request
    # exc: The exception that was raised
    # Returns: A Response object
    pass

Or synchronous:

def handler(request: Request, exc: Exception) -> Response:
    pass

Registering Handlers ¶

During app initialization:

app = Stario(
    exception_handlers={
        404: custom_404,
        Exception: catch_all_handler,
        ValueError: value_error_handler,
    }
)

After initialization:

app.add_exception_handler(404, custom_404)
app.add_exception_handler(ValueError, value_error_handler)

Status Code Handlers ¶

Map status codes to handlers:

exception_handlers={
    404: not_found_handler,
    500: internal_error_handler,
    403: forbidden_handler,
}

Exception Type Handlers ¶

Map exception classes to handlers:

class UserNotFoundError(Exception):
    pass

class PermissionDeniedError(Exception):
    pass

async def user_not_found_handler(request: Request, exc: UserNotFoundError):
    return HTMLResponse("User not found", status_code=404)

async def permission_denied_handler(request: Request, exc: PermissionDeniedError):
    return HTMLResponse("Permission denied", status_code=403)

exception_handlers={
    UserNotFoundError: user_not_found_handler,
    PermissionDeniedError: permission_denied_handler,
}

Catch-All Handler ¶

Handle any exception type:

async def catch_all(request: Request, exc: Exception):
    # Logs are already handled by GuardianMiddleware
    return HTMLResponse(
        content="An unexpected error occurred",
        status_code=500,
    )

exception_handlers={
    Exception: catch_all,  # Catches everything
}

Creating Domain Exceptions ¶

Define your own exception hierarchy for domain errors:

class AppException(Exception):
    """Base exception for all app exceptions"""
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code
        super().__init__(message)

class UserNotFoundError(AppException):
    def __init__(self, user_id: int):
        super().__init__(
            f"User {user_id} not found",
            status_code=404,
        )

class InsufficientPermissionsError(AppException):
    def __init__(self, required_permission: str):
        super().__init__(
            f"Requires permission: {required_permission}",
            status_code=403,
        )

# Generic handler for all app exceptions
async def app_exception_handler(request: Request, exc: AppException):
    return HTMLResponse(
        content=render(div(h1("Error"), p(exc.message))),
        status_code=exc.status_code,
    )

app = Stario(
    exception_handlers={
        AppException: app_exception_handler,
    }
)

# Use in handlers
async def get_user(user_id: PathParam[int], db: Database):
    user = await db.get_user(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return render_user(user)

Exceptions in Dependencies ¶

Exceptions in dependencies propagate the same way as handler exceptions:

async def get_current_user(request: Request, db: Database) -> User:
    user_id = request.session.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Not authenticated")

    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid session")

    return user

async def protected_handler(user: Annotated[User, get_current_user]):
    # If get_current_user raises, handler never runs
    # Exception is caught by middleware
    return div(f"Hello, {user.name}")

Logging Exceptions ¶

All exceptions are automatically logged by GuardianMiddleware with:

• Request ID (request ID) - for tracing
• Stack trace - for debugging
• Request details (method, path, client IP)
• Response status code
• Whether response had already started (streaming errors)

The request ID allows you to search logs and see the entire request lifecycle, including what led to the exception.

See Logging for details on how exception logging works.

Testing Exception Handlers ¶

Use Starlette's test client to test exception handlers:

from starlette.testclient import TestClient

def test_404_handler():
    client = TestClient(app)
    response = client.get("/nonexistent")
    assert response.status_code == 404
    assert "Page Not Found" in response.text

def test_custom_exception():
    client = TestClient(app)
    response = client.get("/user/99999")  # Nonexistent user
    assert response.status_code == 404
    assert "User 99999 not found" in response.text

def test_streaming_exception():
    client = TestClient(app)
    # Streaming errors return 200 (status already sent)
    response = client.get("/stream")
    assert response.status_code == 200
    # But error content is in the stream
    assert "error" in response.text.lower()

See Testing for comprehensive testing documentation including exception handler tests.

Best Practices ¶

1. Validate early in regular handlers—raise before streaming starts
2. In generators, validate everything before the first yield
3. Use domain exceptions for business logic errors
4. Use HTTPException for HTTP-level errors (401, 403, 404)
5. Create custom handlers for better UX on errors
6. Don't catch Exception broadly in handlers—let middleware handle it
7. Log important context before raising exceptions
8. Use request IDs to trace requests through logs

Next Steps ¶

• Learn about Middlewares and exception flow
• Understand Logging to see how exceptions are logged with request IDs
• Explore Endpoints for generator/streaming behavior
• See Testing for testing exception handlers

See Also ¶

• Logging - How exceptions are logged with request IDs
• Middlewares - GuardianMiddleware exception handling
• Streaming Responses - Exception handling in generators