Skip to content

Database connection management plugin for Litestar and asyncpg

License

Notifications You must be signed in to change notification settings

litestar-org/litestar-asyncpg

Repository files navigation

Litestar Logo - Light Litestar Logo - Dark

Project Status
CI/CD Latest Release ci Documentation Building
Quality Coverage Quality Gate Status Maintainability Rating Reliability Rating Security Rating
Package PyPI - Version PyPI - Support Python Versions Advanced Alchemy PyPI - Downloads
Community Reddit Discord Matrix Medium Twitter Blog
Meta Litestar Project types - Mypy License - MIT Litestar Sponsors linting - Ruff code style - Ruff
# Litestar Asyncpg

A barebones AsyncPG plugin for Litestar. This plugin is useful for when you plan to use no ORM or need to manage the postgres connection separately.

Usage

Installation

pip install litestar-asyncpg

Example

Here is a basic application that demonstrates how to use the plugin.

from __future__ import annotations

from typing import TYPE_CHECKING

from litestar import Controller, Litestar, get
from litestar_asyncpg import AsyncpgConfig, AsyncpgPlugin, PoolConfig

if TYPE_CHECKING:
    from asyncpg import Connection


class SampleController(Controller):
    @get(path="/sample")
    async def sample_route(self, db_connection: Connection) -> dict[str, str]:
        """Check database available and returns app config info."""
        result = await db_connection.fetch("select 1")
        return {"select_1": str(result)}


asyncpg = AsyncpgPlugin(config=AsyncpgConfig(pool_config=PoolConfig(dsn="postgresql://app:app@localhost:5432/app")))
app = Litestar(plugins=[asyncpg], route_handlers=[SampleController])