Application

The main application class and lifecycle management.

Xitzin

The core application class that handles routing, middleware, and server lifecycle.

Xitzin

Xitzin(
    *,
    title: str = "Xitzin App",
    version: str = "0.1.0",
    templates_dir: Path | str | None = None,
)

Gemini Application Framework.

Xitzin provides an interface for building Gemini applications.

Example

app = Xitzin(title="My Capsule")

@app.gemini("/") def homepage(request: Request): return "# Welcome to my capsule!"

@app.gemini("/user/{username}") def profile(request: Request, username: str): return f"# {username}'s Profile"

if name == "main": app.run()

Create a new Xitzin application.

Parameters:

Name	Type	Description	Default
title	str	Application title (for documentation).	'Xitzin App'
version	str	Application version.	'0.1.0'
templates_dir	Path | str | None	Directory containing Gemtext templates.	None

Source code in src/xitzin/application.py

def __init__(
    self,
    *,
    title: str = "Xitzin App",
    version: str = "0.1.0",
    templates_dir: Path | str | None = None,
) -> None:
    """Create a new Xitzin application.

    Args:
        title: Application title (for documentation).
        version: Application version.
        templates_dir: Directory containing Gemtext templates.
    """
    self.title = title
    self.version = version
    self._router = Router()
    self._state = AppState()
    self._templates: TemplateEngine | None = None
    self._startup_handlers: list[Callable[[], Any]] = []
    self._shutdown_handlers: list[Callable[[], Any]] = []
    self._middleware: list[Callable[..., Any]] = []
    self._tasks: list[BackgroundTask] = []
    self._task_handles: list[asyncio.Task[Any]] = []
    self._sub_apps: list[Xitzin] = []

    if templates_dir:
        self._init_templates(Path(templates_dir))

state property

state: AppState

Application-level state storage.

template

template(name: str, **context: Any) -> Any

Render a template.

Parameters:

Name	Type	Description	Default
name	str	Template filename (e.g., "page.gmi").	required
**context	Any	Variables to pass to the template.	{}

Returns:

Type	Description
Any	TemplateResponse that can be returned from handlers.

Raises:

Type	Description
RuntimeError	If no templates directory was configured.

Source code in src/xitzin/application.py

def template(self, name: str, **context: Any) -> Any:
    """Render a template.

    Args:
        name: Template filename (e.g., "page.gmi").
        **context: Variables to pass to the template.

    Returns:
        TemplateResponse that can be returned from handlers.

    Raises:
        RuntimeError: If no templates directory was configured.
    """
    if self._templates is None:
        msg = "No templates directory configured"
        raise RuntimeError(msg)
    return self._templates.render(name, **context)

reverse

reverse(name: str, **params: Any) -> str

Build URL for a named route.

Parameters:

Name	Type	Description	Default
name	str	Route name.	required
**params	Any	Path parameters.	{}

Returns:

Type	Description
str	URL path string.

Raises:

Type	Description
ValueError	If route name not found or parameters missing.

Example

url = app.reverse("user_profile", username="alice")

Returns "/user/alice"

Source code in src/xitzin/application.py

def reverse(self, name: str, **params: Any) -> str:
    """Build URL for a named route.

    Args:
        name: Route name.
        **params: Path parameters.

    Returns:
        URL path string.

    Raises:
        ValueError: If route name not found or parameters missing.

    Example:
        url = app.reverse("user_profile", username="alice")
        # Returns "/user/alice"
    """
    return self._router.reverse(name, **params)

redirect

redirect(
    name: str, *, permanent: bool = False, **params: Any
) -> Redirect

Create a redirect to a named route.

Parameters:

Name	Type	Description	Default
name	str	Route name.	required
permanent	bool	If True, use status 31 (permanent redirect).	False
**params	Any	Path parameters.	{}

Returns:

Type	Description
Redirect	Redirect response object.

Example

@app.gemini("/old-profile/{username}") def old_profile(request: Request, username: str): return app.redirect("user_profile", username=username, permanent=True)

Source code in src/xitzin/application.py

def redirect(
    self, name: str, *, permanent: bool = False, **params: Any
) -> Redirect:
    """Create a redirect to a named route.

    Args:
        name: Route name.
        permanent: If True, use status 31 (permanent redirect).
        **params: Path parameters.

    Returns:
        Redirect response object.

    Example:
        @app.gemini("/old-profile/{username}")
        def old_profile(request: Request, username: str):
            return app.redirect("user_profile", username=username, permanent=True)
    """
    url = self.reverse(name, **params)
    return Redirect(url, permanent=permanent)

gemini

gemini(
    path: str, *, name: str | None = None
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a route handler.

Parameters:

Name	Type	Description	Default
path	str	URL path pattern (e.g., "/user/{id}").	required
name	str | None	Optional route name for URL reversing. Defaults to function name.	None

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[..., Any]]	Decorator function.

Example

@app.gemini("/") def home(request: Request): return "# Home"

@app.gemini("/user/{username}", name="user_profile") def profile(request: Request, username: str): return f"# {username}"

Source code in src/xitzin/application.py

def gemini(
    self, path: str, *, name: str | None = None
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register a route handler.

    Args:
        path: URL path pattern (e.g., "/user/{id}").
        name: Optional route name for URL reversing. Defaults to function name.

    Returns:
        Decorator function.

    Example:
        @app.gemini("/")
        def home(request: Request):
            return "# Home"

        @app.gemini("/user/{username}", name="user_profile")
        def profile(request: Request, username: str):
            return f"# {username}"
    """

    def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
        route = Route(path, handler, name=name)
        self._router.add_route(route)
        return handler

    return decorator

input

input(
    path: str,
    *,
    prompt: str,
    sensitive: bool = False,
    name: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register an input route (status 10/11 flow).

When a request arrives without a query string, the client is prompted for input. When the request includes a query string, the handler is called with the decoded input as the query parameter.

Parameters:

Name	Type	Description	Default
path	str	URL path pattern.	required
prompt	str	Prompt text shown to the user.	required
sensitive	bool	If True, use status 11 (sensitive input).	False
name	str | None	Optional route name for URL reversing. Defaults to function name.	None

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[..., Any]]	Decorator function.

Example

@app.input("/search", prompt="Enter search query:", name="search") def search(request: Request, query: str): return f"# Results for: {query}"

Source code in src/xitzin/application.py

def input(
    self,
    path: str,
    *,
    prompt: str,
    sensitive: bool = False,
    name: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register an input route (status 10/11 flow).

    When a request arrives without a query string, the client is prompted
    for input. When the request includes a query string, the handler is
    called with the decoded input as the `query` parameter.

    Args:
        path: URL path pattern.
        prompt: Prompt text shown to the user.
        sensitive: If True, use status 11 (sensitive input).
        name: Optional route name for URL reversing. Defaults to function name.

    Returns:
        Decorator function.

    Example:
        @app.input("/search", prompt="Enter search query:", name="search")
        def search(request: Request, query: str):
            return f"# Results for: {query}"
    """

    def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
        route = Route(
            path, handler, name=name, input_prompt=prompt, sensitive_input=sensitive
        )
        self._router.add_route(route)
        return handler

    return decorator

titan

titan(
    path: str,
    *,
    name: str | None = None,
    auth_tokens: list[str] | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a Titan upload handler.

Titan is the upload companion protocol to Gemini. This decorator registers a handler for Titan upload requests.

Parameters:

Name	Type	Description	Default
path	str	URL path pattern (e.g., "/upload/{filename}").	required
name	str | None	Optional route name. Defaults to function name.	None
auth_tokens	list[str] | None	List of valid authentication tokens. If provided, requests without a valid token are rejected with status 60.	None

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[..., Any]]	Decorator function.

Example

@app.titan("/upload/{filename}", auth_tokens=["secret123"]) def upload(request: TitanRequest, content: bytes, mime_type: str, token: str | None, filename: str): if request.is_delete(): Path(f"./uploads/{filename}").unlink() return "# Deleted" Path(f"./uploads/{filename}").write_bytes(content) return "# Upload successful"

Source code in src/xitzin/application.py

def titan(
    self,
    path: str,
    *,
    name: str | None = None,
    auth_tokens: list[str] | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register a Titan upload handler.

    Titan is the upload companion protocol to Gemini. This decorator
    registers a handler for Titan upload requests.

    Args:
        path: URL path pattern (e.g., "/upload/{filename}").
        name: Optional route name. Defaults to function name.
        auth_tokens: List of valid authentication tokens. If provided,
            requests without a valid token are rejected with status 60.

    Returns:
        Decorator function.

    Example:
        @app.titan("/upload/{filename}", auth_tokens=["secret123"])
        def upload(request: TitanRequest, content: bytes,
                   mime_type: str, token: str | None, filename: str):
            if request.is_delete():
                Path(f"./uploads/{filename}").unlink()
                return "# Deleted"
            Path(f"./uploads/{filename}").write_bytes(content)
            return "# Upload successful"
    """

    def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
        route = TitanRoute(path, handler, name=name, auth_tokens=auth_tokens)
        self._router.add_titan_route(route)
        return handler

    return decorator

mount

mount(
    path: str,
    handler: Callable[..., Any],
    *,
    name: str | None = None,
) -> None

Mount a handler at a path prefix.

Mounted handlers receive requests for any path starting with the prefix. The handler receives (request, path_info) where path_info is the remaining path after the mount prefix.

Parameters:

Name	Type	Description	Default
path	str	Mount point prefix (e.g., "/cgi-bin", "/api").	required
handler	Callable[..., Any]	Callable that takes (request, path_info) and returns a response.	required
name	str | None	Optional name for the mount.	None

Example

from xitzin.cgi import CGIHandler

app.mount("/cgi-bin", CGIHandler(script_dir="./scripts"))

Requests to /cgi-bin/hello.py will call:

handler(request, path_info="/hello.py")

Source code in src/xitzin/application.py

def mount(
    self,
    path: str,
    handler: Callable[..., Any],
    *,
    name: str | None = None,
) -> None:
    """Mount a handler at a path prefix.

    Mounted handlers receive requests for any path starting with the prefix.
    The handler receives (request, path_info) where path_info is the
    remaining path after the mount prefix.

    Args:
        path: Mount point prefix (e.g., "/cgi-bin", "/api").
        handler: Callable that takes (request, path_info) and returns a response.
        name: Optional name for the mount.

    Example:
        from xitzin.cgi import CGIHandler

        app.mount("/cgi-bin", CGIHandler(script_dir="./scripts"))

        # Requests to /cgi-bin/hello.py will call:
        # handler(request, path_info="/hello.py")
    """
    mounted = MountedRoute(path, handler, name=name)
    self._router.add_mounted_route(mounted)

vhost

vhost(
    hosts: dict[str, "Xitzin"],
    *,
    default_app: "Xitzin | None" = None,
    fallback_status: int = 53,
    fallback_handler: Callable[[Request], Any]
    | None = None,
) -> None

Configure virtual hosting for this application.

This is a convenience method that creates and registers VirtualHostMiddleware. The middleware routes requests to different apps based on hostname.

Parameters:

Name	Type	Description	Default
hosts	dict[str, 'Xitzin']	Mapping of hostname patterns to Xitzin apps. Patterns can be exact ("example.com") or wildcards ("*.example.com"). Exact matches are checked first, then wildcards in definition order.	required
default_app	'Xitzin | None'	Default app when no pattern matches.	None
fallback_status	int	Status code for unmatched hosts (default: 53). Common values: 53 (Proxy Refused), 51 (Not Found), 59 (Bad Request).	53
fallback_handler	Callable[[Request], Any] | None	Custom handler for unmatched hosts. Takes precedence over default_app and fallback_status.	None

Example

main_app = Xitzin(title="Main") blog_app = Xitzin(title="Blog") api_app = Xitzin(title="API")

@blog_app.gemini("/") def blog_home(request: Request): return "# Blog Home"

@api_app.gemini("/") def api_home(request: Request): return "# API Home"

@main_app.gemini("/") def main_home(request: Request): return "# Main Home"

Configure as gateway

main_app.vhost({ "blog.example.com": blog_app, "*.api.example.com": api_app, }, default_app=main_app)

main_app.run()

Source code in src/xitzin/application.py

def vhost(
    self,
    hosts: dict[str, "Xitzin"],
    *,
    default_app: "Xitzin | None" = None,
    fallback_status: int = 53,
    fallback_handler: Callable[[Request], Any] | None = None,
) -> None:
    """Configure virtual hosting for this application.

    This is a convenience method that creates and registers VirtualHostMiddleware.
    The middleware routes requests to different apps based on hostname.

    Args:
        hosts: Mapping of hostname patterns to Xitzin apps.
            Patterns can be exact ("example.com") or wildcards ("*.example.com").
            Exact matches are checked first, then wildcards in definition order.
        default_app: Default app when no pattern matches.
        fallback_status: Status code for unmatched hosts (default: 53).
            Common values: 53 (Proxy Refused), 51 (Not Found), 59 (Bad Request).
        fallback_handler: Custom handler for unmatched hosts.
            Takes precedence over default_app and fallback_status.

    Example:
        main_app = Xitzin(title="Main")
        blog_app = Xitzin(title="Blog")
        api_app = Xitzin(title="API")

        @blog_app.gemini("/")
        def blog_home(request: Request):
            return "# Blog Home"

        @api_app.gemini("/")
        def api_home(request: Request):
            return "# API Home"

        @main_app.gemini("/")
        def main_home(request: Request):
            return "# Main Home"

        # Configure as gateway
        main_app.vhost({
            "blog.example.com": blog_app,
            "*.api.example.com": api_app,
        }, default_app=main_app)

        main_app.run()
    """
    from .middleware import VirtualHostMiddleware

    vhost_mw = VirtualHostMiddleware(
        hosts,
        default_app=default_app,
        fallback_status=fallback_status,
        fallback_handler=fallback_handler,
    )

    @self.middleware
    async def virtual_host_dispatcher(
        request: Request, call_next: Callable[..., Any]
    ) -> GeminiResponse:
        return await vhost_mw(request, call_next)

    # Track sub-apps for lifecycle management
    for app in hosts.values():
        if app is not self and app not in self._sub_apps:
            self._sub_apps.append(app)
    if (
        default_app is not None
        and default_app is not self
        and default_app not in self._sub_apps
    ):
        self._sub_apps.append(default_app)

task

task(
    *,
    interval: str | int | float | None = None,
    cron: str | None = None,
) -> Callable[[Callable[[], Any]], Callable[[], Any]]

Register a background task.

Tasks run continuously while the server is running. They are started after startup handlers and stopped before shutdown handlers.

Parameters:

Name	Type	Description	Default
interval	str | int | float | None	Run every N seconds (int) or duration string ("1h", "30m", "1d").	None
cron	str | None	Cron expression string ("0 * * * *" runs hourly). Requires croniter: pip install 'xitzin[tasks]'	None

Exactly one of interval or cron must be provided.

Returns:

Type	Description
Callable[[Callable[[], Any]], Callable[[], Any]]	Decorator function.

Raises:

Type	Description
TaskConfigurationError	If neither or both parameters provided, or if cron is used but croniter is not installed.

Example

@app.task(interval="1h") async def cleanup(): await app.state.db.cleanup_old_records()

@app.task(cron="0 2 * * *") # 2 AM daily def backup(): backup_database()

Source code in src/xitzin/application.py

def task(
    self,
    *,
    interval: str | int | float | None = None,
    cron: str | None = None,
) -> Callable[[Callable[[], Any]], Callable[[], Any]]:
    """Register a background task.

    Tasks run continuously while the server is running. They are started
    after startup handlers and stopped before shutdown handlers.

    Args:
        interval: Run every N seconds (int) or duration string ("1h", "30m", "1d").
        cron: Cron expression string ("0 * * * *" runs hourly).
            Requires croniter: pip install 'xitzin[tasks]'

    Exactly one of interval or cron must be provided.

    Returns:
        Decorator function.

    Raises:
        TaskConfigurationError: If neither or both parameters provided,
            or if cron is used but croniter is not installed.

    Example:
        @app.task(interval="1h")
        async def cleanup():
            await app.state.db.cleanup_old_records()

        @app.task(cron="0 2 * * *")  # 2 AM daily
        def backup():
            backup_database()
    """
    from .tasks import BackgroundTask, parse_interval

    # Validate parameters
    if interval is None and cron is None:
        raise TaskConfigurationError("Either 'interval' or 'cron' must be provided")
    if interval is not None and cron is not None:
        raise TaskConfigurationError(
            "Only one of 'interval' or 'cron' can be provided, not both"
        )

    # Check croniter availability
    if cron is not None:
        try:
            from croniter import croniter as _  # noqa: F401
        except ImportError:
            raise TaskConfigurationError(
                "croniter is required for cron tasks. "
                "Install with: pip install 'xitzin[tasks]'"
            ) from None

    def decorator(handler: Callable[[], Any]) -> Callable[[], Any]:
        # Parse interval if provided
        parsed_interval = parse_interval(interval) if interval else None

        task = BackgroundTask(
            handler=handler,
            interval=parsed_interval,
            cron=cron,
            name=getattr(handler, "__name__", "<anonymous>"),
        )
        self._tasks.append(task)
        return handler

    return decorator

on_startup

on_startup(handler: Callable[[], Any]) -> Callable[[], Any]

Register a startup event handler.

Parameters:

Name	Type	Description	Default
handler	Callable[[], Any]	Function to call on startup.	required

Example

@app.on_startup async def startup(): app.state.db = await create_db_pool()

Source code in src/xitzin/application.py

def on_startup(self, handler: Callable[[], Any]) -> Callable[[], Any]:
    """Register a startup event handler.

    Args:
        handler: Function to call on startup.

    Example:
        @app.on_startup
        async def startup():
            app.state.db = await create_db_pool()
    """
    self._startup_handlers.append(handler)
    return handler

on_shutdown

on_shutdown(
    handler: Callable[[], Any],
) -> Callable[[], Any]

Register a shutdown event handler.

Parameters:

Name	Type	Description	Default
handler	Callable[[], Any]	Function to call on shutdown.	required

Example

@app.on_shutdown async def shutdown(): await app.state.db.close()

Source code in src/xitzin/application.py

def on_shutdown(self, handler: Callable[[], Any]) -> Callable[[], Any]:
    """Register a shutdown event handler.

    Args:
        handler: Function to call on shutdown.

    Example:
        @app.on_shutdown
        async def shutdown():
            await app.state.db.close()
    """
    self._shutdown_handlers.append(handler)
    return handler

middleware

middleware(
    handler: Callable[..., Any],
) -> Callable[..., Any]

Register middleware as a decorator.

Middleware receives (request, call_next) and must call call_next to continue processing.

Parameters:

Name	Type	Description	Default
handler	Callable[..., Any]	Middleware function.	required

Example

@app.middleware async def log_requests(request: Request, call_next): print(f"Request: {request.path}") response = await call_next(request) print(f"Response: {response.status}") return response

Source code in src/xitzin/application.py

def middleware(self, handler: Callable[..., Any]) -> Callable[..., Any]:
    """Register middleware as a decorator.

    Middleware receives (request, call_next) and must call call_next
    to continue processing.

    Args:
        handler: Middleware function.

    Example:
        @app.middleware
        async def log_requests(request: Request, call_next):
            print(f"Request: {request.path}")
            response = await call_next(request)
            print(f"Response: {response.status}")
            return response
    """
    self._middleware.append(handler)
    return handler

run

run(
    host: str = "localhost",
    port: int = 1965,
    certfile: Path | str | None = None,
    keyfile: Path | str | None = None,
) -> None

Run the server (blocking).

Parameters:

Name	Type	Description	Default
host	str	Host address to bind to.	'localhost'
port	int	Port to bind to.	1965
certfile	Path | str | None	Path to TLS certificate file.	None
keyfile	Path | str | None	Path to TLS private key file.	None

Source code in src/xitzin/application.py

def run(
    self,
    host: str = "localhost",
    port: int = 1965,
    certfile: Path | str | None = None,
    keyfile: Path | str | None = None,
) -> None:
    """Run the server (blocking).

    Args:
        host: Host address to bind to.
        port: Port to bind to.
        certfile: Path to TLS certificate file.
        keyfile: Path to TLS private key file.
    """
    try:
        asyncio.run(
            self.run_async(host=host, port=port, certfile=certfile, keyfile=keyfile)
        )
    except KeyboardInterrupt:
        print("\n[Xitzin] Shutting down...")

run_async async

run_async(
    host: str = "localhost",
    port: int = 1965,
    certfile: Path | str | None = None,
    keyfile: Path | str | None = None,
) -> None

Run the server asynchronously.

Parameters:

Name	Type	Description	Default
host	str	Host address to bind to.	'localhost'
port	int	Port to bind to.	1965
certfile	Path | str | None	Path to TLS certificate file.	None
keyfile	Path | str | None	Path to TLS private key file.	None

Source code in src/xitzin/application.py

async def run_async(
    self,
    host: str = "localhost",
    port: int = 1965,
    certfile: Path | str | None = None,
    keyfile: Path | str | None = None,
) -> None:
    """Run the server asynchronously.

    Args:
        host: Host address to bind to.
        port: Port to bind to.
        certfile: Path to TLS certificate file.
        keyfile: Path to TLS private key file.
    """
    from nauyaca.server.protocol import GeminiServerProtocol
    from nauyaca.server.tls_protocol import TLSServerProtocol
    from nauyaca.security import generate_self_signed_cert  # ty: ignore[unresolved-import]
    from nauyaca.security import create_pyopenssl_server_context  # ty: ignore[unresolved-import]
    import tempfile

    # Run startup handlers
    await self._run_startup()

    # Start background tasks
    await self._run_tasks()

    try:
        # Create PyOpenSSL context (accepts any self-signed client cert)
        if certfile and keyfile:
            ssl_context = create_pyopenssl_server_context(
                str(certfile),
                str(keyfile),
                request_client_cert=True,
            )
        else:
            # Generate self-signed cert for development
            cert_pem, key_pem = generate_self_signed_cert(
                hostname="localhost",
                key_size=2048,
                valid_days=365,
            )

            with (
                tempfile.NamedTemporaryFile(
                    suffix=".pem", delete=False, mode="wb"
                ) as cf,
                tempfile.NamedTemporaryFile(
                    suffix=".key", delete=False, mode="wb"
                ) as kf,
            ):
                cf.write(cert_pem)
                kf.write(key_pem)
                cf.flush()
                kf.flush()
                print("[Xitzin] Using self-signed certificate (development only)")
                ssl_context = create_pyopenssl_server_context(
                    cf.name,
                    kf.name,
                    request_client_cert=True,
                )

        # Create handler that routes to our app
        async def handle(request: GeminiRequest) -> GeminiResponse:
            return await self._handle_request(request)

        # Create Titan upload handler if Titan routes are registered
        upload_handler = None
        if self._router.has_titan_routes():
            from nauyaca.server.handler import UploadHandler

            class XitzinUploadHandler(UploadHandler):
                """Wrapper to route Titan uploads to Xitzin handlers."""

                def __init__(self, app: "Xitzin") -> None:
                    self._app = app

                async def handle_upload(
                    self, request: NauyacaTitanRequest
                ) -> GeminiResponse:
                    return await self._app._handle_titan_request(request)

            upload_handler = XitzinUploadHandler(self)
            print("[Xitzin] Titan upload support enabled")

        # Use TLSServerProtocol for manual TLS handling
        # (supports self-signed client certs)
        def create_protocol() -> TLSServerProtocol:
            return TLSServerProtocol(
                lambda: GeminiServerProtocol(handle, None, upload_handler),
                ssl_context,
            )

        loop = asyncio.get_running_loop()
        server = await loop.create_server(
            create_protocol,
            host,
            port,
        )

        print(f"[Xitzin] {self.title} v{self.version}")
        print(f"[Xitzin] Serving at gemini://{host}:{port}/")

        async with server:
            await server.serve_forever()

    finally:
        # Stop background tasks
        await self._stop_tasks()
        await self._run_shutdown()

AppState

Application-level state storage for shared resources like database connections.

AppState

Application-level state storage.

Store shared resources like database connections here.

Example

app.state.db = create_db_connection()