Added some examples for AioHTTP and FastAPI

Author: chandr-andr

docs/examples/aiohttp/start_example.py

@@ -0,0 +1,44 @@
+# Start example
+import asyncio
+from typing import cast
+from aiohttp import web
+from psqlpy import PSQLPool
+
+
+async def start_db_pool(app: web.Application) -> None:
+    """Initialize database connection pool."""
+    db_pool = PSQLPool(
+        dsn="postgres://postgres:postgres@localhost:5432/postgres",
+        max_db_pool_size=2,
+    )
+    await db_pool.startup()
+
+    app["db_pool"] = db_pool
+
+
+async def stop_db_pool(app: web.Application) -> None:
+    """Close database connection pool."""
+    db_pool = cast(PSQLPool, app.db_pool)
+    await db_pool.close()
+
+
+async def pg_pool_example(request: web.Request):
+    db_pool = cast(PSQLPool, request.app["db_pool"])
+    connection = await db_pool.connection()
+    await asyncio.sleep(10)
+    query_result = await connection.execute(
+        "SELECT * FROM users",
+    )
+    dict_result = query_result.result()
+    return web.json_response(
+        data=dict_result,
+    )
+
+
+application = web.Application()
+application.on_startup.append(start_db_pool)
+application.add_routes([web.get('/', pg_pool_example)])
+
+
+if __name__ == "__main__":
+    web.run_app(application)

docs/examples/fastapi/advanced_example.py

@@ -0,0 +1,50 @@
+# Start example
+import asyncio
+from contextlib import asynccontextmanager
+from typing import Annotated, AsyncGenerator, cast
+from fastapi import Depends, FastAPI, Request
+from fastapi.responses import JSONResponse
+from psqlpy import PSQLPool, Connection
+import uvicorn
+
+
+db_pool = PSQLPool(
+    dsn="postgres://postgres:postgres@localhost:5432/postgres",
+    max_db_pool_size=2,
+)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
+    """Startup database connection pool and close it on shutdown."""
+    await db_pool.startup()
+    app.state.db_pool = db_pool
+    yield
+    await db_pool.close()
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+async def some_long_func() -> None:
+    # Some very long execution.
+    print("Executing...")
+    await asyncio.sleep(10)
+    print("Done.")
+
+
+@app.get("/")
+async def pg_pool_example():
+    await some_long_func()
+    db_connection = await db_pool.connection()
+    query_result = await db_connection.execute(
+        "SELECT * FROM users",
+    )
+    return JSONResponse(content=query_result.result())
+
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "advanced_example:app",
+        port=8001,
+    )

docs/examples/fastapi/start_example.py

@@ -0,0 +1,44 @@
+# Start example
+from contextlib import asynccontextmanager
+from typing import Annotated, AsyncGenerator, cast
+from fastapi import Depends, FastAPI, Request
+from fastapi.responses import JSONResponse
+from psqlpy import PSQLPool, Connection
+import uvicorn
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
+    """Startup database connection pool and close it on shutdown."""
+    db_pool = PSQLPool(
+        dsn="postgres://postgres:postgres@localhost:5432/postgres",
+        max_db_pool_size=2,
+    )
+    await db_pool.startup()
+    app.state.db_pool = db_pool
+    yield
+    await db_pool.close()
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+async def db_connection(request: Request) -> Connection:
+    """Retrieve new connection from connection pool and return it."""
+    return await (cast(PSQLPool, request.app.state.db_pool)).connection()
+
+
+@app.get("/")
+async def pg_pool_example(
+    db_connection: Annotated[Connection, Depends(db_connection)],
+):
+    query_result = await db_connection.execute(
+        "SELECT * FROM users",
+    )
+    return JSONResponse(content=query_result.result())
+
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "start_example:app",
+    )