feat: provide connection outside of Litestar application

litestar_asyncpg/config.py

@@ -223,3 +223,18 @@ async def provide_connection(
             async with pool.acquire() as connection:
                 set_scope_state(scope, CONNECTION_SCOPE_KEY, connection)
                 yield connection
+
+    @asynccontextmanager
+    async def get_connection(
+        self,
+    ) -> AsyncGenerator[Union[PoolConnectionProxy,Connection], None]:  # noqa: UP007
+        """Create a connection instance.
+
+        Args:
+            pool: The pool to grab a connection from
+
+        Returns:
+            A connection instance.
+        """
+        async with (await self.create_pool()).acquire() as connection:
+            yield connection

tests/test_config.py

@@ -0,0 +1,16 @@
+
+
+import pytest
+
+from litestar_asyncpg import AsyncpgConfig, PoolConfig
+
+pytestmark = pytest.mark.anyio
+
+
+async def test_get_connection(postgres_service: None, postgres_docker_ip: str,  postgres_user: str, postgres_password: str, postgres_database: str, postgres_port:int) -> None:
+    asyncpg_config = AsyncpgConfig(pool_config=PoolConfig(dsn=f"postgresql://{postgres_user}:{postgres_password}@{postgres_docker_ip}:{postgres_port}/{postgres_database}"))
+
+
+    async with asyncpg_config.get_connection() as db_connection:
+        r = await db_connection.fetch("select 1 as one")
+        assert r[0]["one"] == 1