task: update docs

.github/workflows/run-tests.yml

@@ -5,6 +5,7 @@ on:
     branches:
       - 'fix/**'
       - 'feat/**'
+      - 'task/**'
 jobs:
   Run-Tests:
     runs-on: ubuntu-latest
@@ -24,4 +25,4 @@ jobs:
       - name: Install dependencies
         run: poetry install
       - name: Run tests
-        run: poetry run pytest
+        run: poetry run pytest --capture=no

README.md

@@ -1,3 +1,3 @@
 # fastapi-boilerplate
 
-Kickstart your [FastAPI](https://fastapi.tiangolo.com/) development with ease by forking this boilerplate repository.
+Kickstart your [FastAPI](https://fastapi.tiangolo.com/) development with ease.

pyproject.toml

@@ -12,6 +12,7 @@ homepage = "https://svaponi.github.io/fastapi-boilerplate"
 python = "^3.12"
 fastapi = "^0.104.1"
 httpx = "^0.27.0"
+asyncpg = "^0.29.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.1.1"
@@ -20,6 +21,7 @@ python-dotenv = "^1.0.1"
 uvicorn = "^0.29.0"
 black = "^24.4.0"
 pytest-httpserver = "^1.0.10"
+testcontainers = "^4.7.2"
 
 [build-system]
 requires = ["poetry-core"]

src/app/api/users.py

@@ -0,0 +1,38 @@
+import fastapi
+
+from app.datalayer.model import UserAccount
+from app.datalayer.users import UserService
+
+router = fastapi.APIRouter()
+
+
+@router.get("")
+async def get_users(
+    data_service: UserService = fastapi.Depends(),
+    page: int = 0,
+    size: int = 10,
+) -> list[UserAccount]:
+    # Read users
+    users = await data_service.read_users()
+
+    # Create users
+    if not users:
+        await data_service.create_user("Alice", "alice@example.com")
+        await data_service.create_user("Bob", "bob@example.com")
+
+    # Read users
+    users = await data_service.read_users()
+
+    # # Update user
+    # user_id = users[0].id
+    # await data_service.update_user(user_id, "John")
+    #
+    # # Read users again
+    # await data_service.read_users()
+    #
+    # # Delete user
+    # await data_service.delete_user(2)
+
+    # Read users again
+
+    return users

src/app/api/v1.py

@@ -0,0 +1,9 @@
+import fastapi
+
+from app.api import users
+
+
+def setup_api_v1(app: fastapi.FastAPI):
+    api_router = fastapi.APIRouter()
+    api_router.include_router(users.router, prefix="/users", tags=["users"])
+    app.include_router(api_router, prefix="/api/v1")

src/app/app.py

@@ -0,0 +1,69 @@
+import contextlib
+import logging
+import typing
+
+import fastapi
+from starlette.responses import RedirectResponse
+
+from app.api.v1 import setup_api_v1
+from app.core.cors import setup_cors
+from app.core.datasource import setup_datasource
+from app.core.error_handlers import setup_error_handlers
+from app.core.logging import setup_logging
+from app.core.migration import run_migrations
+from app.core.request_context import setup_request_context
+
+
+# See https://fastapi.tiangolo.com/advanced/events/#lifespan
+@contextlib.asynccontextmanager
+async def _lifespan(app: "App"):
+    app.logger.info(f"Starting ...")
+    await app.datasource.connect()
+    await run_migrations(app.datasource)
+    # ...other startup code
+    app.logger.info("✅ Started")
+    yield
+    app.logger.info("Shutting down ...")
+    await app.datasource.disconnect()
+    # ...other shutdown code
+    app.logger.info("🛑 Shutdown")
+
+
+class App(fastapi.FastAPI):
+    def __init__(
+        self,
+        **extra: typing.Any,
+    ) -> None:
+        # all logs previous to calling setup_logging will be not formatted
+        setup_logging()
+        self.logger = logging.getLogger(f"app")
+
+        super().__init__(
+            lifespan=_lifespan,
+            **extra,
+        )
+
+        # RequestContext is a nice-to-have util that allows to access request related attributes anywhere in the code
+        setup_request_context(self)
+
+        # Http error handlers (equivalent to try block that can handle uncaught exceptions)
+        setup_error_handlers(self)
+
+        # Setup cors
+        setup_cors(self)
+
+        # Setup API routes
+        setup_api_v1(self)
+
+        # Setup DB data source
+        self.datasource = setup_datasource(self)
+
+        if hasattr(self, "docs_url") and self.docs_url:
+
+            @self.get("/", include_in_schema=False)
+            def root():
+                return RedirectResponse(self.docs_url)
+
+
+def create_app():
+    return App()

src/app/core/datasource.py

@@ -0,0 +1,39 @@
+import logging
+import os
+
+import asyncpg
+import fastapi
+
+
+# Simple class that takes care of setting up and tearing down the connection pool.
+# Useful to decouple for the actual connection pool object.
+# It's meant to be used in FastAPI lifespan (see https://fastapi.tiangolo.com/advanced/events/#lifespan).
+class DataSource:
+    def __init__(self, app: fastapi.FastAPI):
+        self._pool: asyncpg.Pool | None = None
+        self.logger = logging.getLogger(__name__)
+
+    async def connect(self):
+        if not self._pool:
+            postgres_url = os.getenv("POSTGRES_URL")
+            assert postgres_url, "missing POSTGRES_URL"
+            self._pool = await asyncpg.create_pool(dsn=postgres_url)
+            self.logger.info("DataSource created")
+
+    async def disconnect(self):
+        if self._pool:
+            await self._pool.close()
+            self.logger.info("DataSource closed")
+
+    @property
+    def pool(self) -> asyncpg.Pool:
+        if not self._pool:
+            raise RuntimeError("DataSource not initialized, you need to call `await connect()` in the lifespan event, "
+                               "see see https://fastapi.tiangolo.com/advanced/events/#lifespan.")
+        return self._pool
+
+
+def setup_datasource(app: fastapi.FastAPI) -> DataSource:
+    return DataSource(app)
+
+

src/app/core/migration.py

@@ -0,0 +1,20 @@
+import os.path
+
+from app.core.datasource import DataSource
+
+
+async def run_migrations(datasource: DataSource):
+    async with datasource.pool.acquire(timeout=2) as connection:
+        await connection.execute("create table if not exists migration (name text primary key, executed_at timestamp)")
+        records = await connection.fetch("select * from migration")
+        print(records)
+        if len(records) > 0:
+            print("Migration table already exists")
+        else:
+            with open(os.path.join(os.path.dirname(__file__), "../../schema.sql")) as f:
+                content = f.read()
+            print(content)
+            await connection.execute(content)
+            await connection.execute("insert into migration (name, executed_at) values ('migration.sql', now())")
+
+

src/app/datalayer/db.py

@@ -0,0 +1,19 @@
+import contextlib
+
+import fastapi
+
+
+class DB:
+    def __init__(self, request: fastapi.Request):
+        self.pool = request.app.datasource.pool
+
+    @contextlib.asynccontextmanager
+    async def transaction(self):
+        async with self.pool.acquire(timeout=2) as connection:
+            async with connection.transaction():
+                yield connection
+
+    @contextlib.asynccontextmanager
+    async def connection(self):
+        async with self.pool.acquire(timeout=2) as connection:
+            yield connection