feat: dependencies override

app/api/dependencies.py

@@ -0,0 +1,8 @@
+from functools import lru_cache
+
+from app.core.settings import Settings
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()  # ty: ignore[missing-argument]

app/api/django/dependencies.py

@@ -1,5 +1,11 @@
+from typing import Annotated
+
+from fast_depends import Depends
+
+from app.api.dependencies import get_settings
 from app.core.django.context import Context
+from app.core.settings import Settings
 
 
-def get_context() -> Context:
-    return Context()
+def get_context(settings: Annotated[Settings, Depends(get_settings)]) -> Context:
+    return Context(settings=settings)

app/api/django/items/routes.py

@@ -80,7 +80,16 @@ def delete(
         return JsonResponse(data="", status=204, safe=False)
 
 
-class ItemViewSpecial(View):
+class DevEnvView(View):
+    @inject
+    def get(self, context: Annotated[Context, Depends(get_context)]) -> JsonResponse:
+        return JsonResponse(
+            data={"environment": context.environment},
+            safe=False,
+        )
+
+
+class DevErrorView(View):
     body_models: ClassVar[dict[str, type[BaseModel]]] = {"POST": ItemCreateError}
 
     @inject(cast=False)

app/api/django/urls.py

@@ -2,12 +2,18 @@
 from django.urls.resolvers import URLPattern, URLResolver
 
 from app.api.django.handlers import custom_handler404
-from app.api.django.items.routes import ItemView, ItemViewDetail, ItemViewSpecial
+from app.api.django.items.routes import (
+    DevEnvView,
+    DevErrorView,
+    ItemView,
+    ItemViewDetail,
+)
 
 urlpatterns: list[URLPattern | URLResolver] = [
     path("items", ItemView.as_view()),
     path("items/<item_id>", ItemViewDetail.as_view()),
-    path("dev/error", ItemViewSpecial.as_view()),
+    path("dev/env", DevEnvView.as_view()),
+    path("dev/error", DevErrorView.as_view()),
 ]
 
 # Override handler to return JSON response

app/api/fastapi/dependencies.py

@@ -1,25 +1,22 @@
 from collections.abc import Iterator
-from functools import lru_cache
 from typing import Annotated
 
 from fastapi import Depends
 from sqlalchemy.orm import Session
 from starlette.requests import Request
 
+from app.api.dependencies import get_settings
 from app.core.settings import Settings
 from app.core.sqlalchemy.context import Context
-from app.infrastructure.sqlalchemy.utils import managed_session
-
-
-@lru_cache
-def get_settings() -> Settings:
-    return Settings()  # ty: ignore[missing-argument]
 
 
 def get_sql_session(request: Request) -> Iterator[Session]:
-    with managed_session(request.app.state.sql_session_factory) as session:
+    with request.app.state.sql_resource.session() as session:
         yield session
 
 
-def get_context(sql_session: Annotated[Session, Depends(get_sql_session)]) -> Context:
-    return Context(sql_session=sql_session)
+def get_context(
+    settings: Annotated[Settings, Depends(get_settings)],
+    sql_session: Annotated[Session, Depends(get_sql_session)],
+) -> Context:
+    return Context(settings=settings, sql_session=sql_session)

app/api/fastapi/dev/router.py

@@ -3,13 +3,21 @@
 from fastapi import APIRouter, Depends, status
 
 from app.api.fastapi.dependencies import get_context
+from app.core.settings import AppEnvironment
 from app.core.sqlalchemy.context import Context
 from app.domain.dev.commands import ItemCreateError, create_item_error_command
 from app.domain.items.entities import Item
 
 router = APIRouter(prefix="/dev")
 
 
+@router.get("/env")
+def get_env(
+    context: Annotated[Context, Depends(get_context)],
+) -> dict[str, AppEnvironment]:
+    return {"environment": context.environment}
+
+
 @router.post("/error", response_model=Item, status_code=status.HTTP_201_CREATED)
 def item_error(
     item_create: ItemCreateError,

app/api/fastapi/lifespan.py

@@ -5,7 +5,7 @@
 
 from app.api.logger import logger
 from app.core.settings import Settings
-from app.infrastructure.sqlalchemy.utils import create_sql_resource
+from app.infrastructure.sqlalchemy.utils import SQLResource
 
 
 def lifespan_factory(
@@ -14,9 +14,8 @@ def lifespan_factory(
 
     @asynccontextmanager
     async def lifespan(app: FastAPI) -> AsyncIterator[None]:
-        sql_resource = create_sql_resource(settings=settings)
-        app.state.sql_engine = sql_resource.engine
-        app.state.sql_session_factory = sql_resource.session_factory
+        sql_resource = SQLResource.from_settings(settings)
+        app.state.sql_resource = sql_resource
         logger.info("Application startup complete")
 
         yield

app/api/flask/dependencies.py

@@ -5,14 +5,18 @@
 from flask import current_app
 from sqlalchemy.orm import Session
 
+from app.api.dependencies import get_settings
+from app.core.settings import Settings
 from app.core.sqlalchemy.context import Context
-from app.infrastructure.sqlalchemy.utils import managed_session
 
 
 def get_sql_session() -> Iterator[Session]:
-    with managed_session(current_app.config["SQL_SESSION_FACTORY"]) as session:
+    with current_app.config["SQL_RESOURCE"].session() as session:
         yield session
 
 
-def get_context(sql_session: Annotated[Session, Depends(get_sql_session)]) -> Context:
-    return Context(sql_session=sql_session)
+def get_context(
+    settings: Annotated[Settings, Depends(get_settings)],
+    sql_session: Annotated[Session, Depends(get_sql_session)],
+) -> Context:
+    return Context(settings=settings, sql_session=sql_session)

app/api/flask/dev/router.py

@@ -1,3 +1,4 @@
+import json
 from typing import Annotated
 
 from fast_depends import Depends, inject
@@ -10,6 +11,16 @@
 router = Blueprint("dev", __name__, url_prefix="/dev")
 
 
+@router.get("/env")
+@inject
+def get_env(context: Annotated[Context, Depends(get_context)]) -> Response:
+    return Response(
+        json.dumps({"environment": context.environment}),
+        status=200,
+        content_type="application/json",
+    )
+
+
 @router.post("/error")
 @inject
 def item_error(

app/api/flask/utils.py

@@ -1,10 +1,9 @@
 from flask import Flask
 
 from app.core.settings import Settings
-from app.infrastructure.sqlalchemy.utils import create_sql_resource
+from app.infrastructure.sqlalchemy.utils import SQLResource
 
 
 def init_app(settings: Settings, app: Flask) -> None:
-    sql_resource = create_sql_resource(settings=settings)
-    app.config["SQL_ENGINE"] = sql_resource.engine
-    app.config["SQL_SESSION_FACTORY"] = sql_resource.session_factory
+    sql_resource = SQLResource.from_settings(settings)
+    app.config["SQL_RESOURCE"] = sql_resource

app/core/django/context.py

@@ -1,11 +1,20 @@
 from functools import cached_property
 
+from app.core.settings import AppEnvironment, Settings
 from app.domain.context import ContextProtocol
 from app.domain.items.repository import ItemRepositoryProtocol
 from app.infrastructure.django.items import ItemRepository
 
 
 class Context(ContextProtocol):
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+
+    @property
+    def environment(self) -> AppEnvironment:
+        # property to test dependencies override
+        return self.settings.environment
+
     @cached_property
     def item_repository(self) -> ItemRepositoryProtocol:
         return ItemRepository()

app/core/settings.py

@@ -1,3 +1,4 @@
+from enum import StrEnum
 from typing import Any
 
 from pydantic import BaseModel, PostgresDsn, SecretStr, computed_field
@@ -7,6 +8,12 @@
 from app.infrastructure.django.apps import DJANGO_APPS
 
 
+class AppEnvironment(StrEnum):
+    DEVELOPMENT = "development"
+    TESTING = "testing"
+    PRODUCTION = "production"
+
+
 class DjangoSettings(BaseModel):
     debug: bool = False
     secret_key: str
@@ -24,6 +31,8 @@ class Settings(BaseSettings):
         nested_model_default_partial_update=True,
     )
 
+    environment: AppEnvironment
+
     django: DjangoSettings
 
     postgres_user: str

app/core/sqlalchemy/context.py

@@ -2,15 +2,22 @@
 
 from sqlalchemy.orm import Session
 
+from app.core.settings import AppEnvironment, Settings
 from app.domain.context import ContextProtocol
 from app.domain.items.repository import ItemRepositoryProtocol
 from app.infrastructure.sqlalchemy.items import SQLItemRepository
 
 
 class Context(ContextProtocol):
-    def __init__(self, sql_session: Session) -> None:
+    def __init__(self, settings: Settings, sql_session: Session) -> None:
+        self.settings = settings
         self.sql_session = sql_session
 
+    @property
+    def environment(self) -> AppEnvironment:
+        # property to test dependencies override
+        return self.settings.environment
+
     @cached_property
     def item_repository(self) -> ItemRepositoryProtocol:
         return SQLItemRepository(session=self.sql_session)

app/infrastructure/sqlalchemy/utils.py

@@ -16,6 +16,38 @@ class SQLResource(BaseModel):
     engine: Engine
     session_factory: sessionmaker[Session]
 
+    @classmethod
+    def from_settings(cls, settings: Settings, /) -> SQLResource:
+        engine = create_engine(
+            url=str(settings.postgres_dsn),
+            **settings.postgres_params,
+        )
+        with engine.connect() as connection:
+            connection.execute(text("SELECT 1"))
+        logger.info("SQL engine up")
+        return cls(
+            engine=engine,
+            session_factory=sessionmaker(bind=engine),
+        )
+
+    def create_all(self) -> None:
+        OrmEntity.metadata.drop_all(self.engine)
+        OrmEntity.metadata.create_all(self.engine)
+
+    @contextmanager
+    def session(self) -> Iterator[Session]:
+        _session = self.session_factory()
+        try:
+            yield _session
+            _session.commit()
+            logger.info("Commit ok")
+        except Exception as error:
+            logger.error(f"Rollback due to {error}")
+            _session.rollback()
+            raise
+        finally:
+            _session.close()
+
     def release(self) -> None:
         logger.info("SQL engine released")
         self.engine.dispose()
@@ -25,32 +57,3 @@ def reset(self) -> None:
             for table in reversed(OrmEntity.metadata.sorted_tables):
                 session.execute(table.delete())
             session.commit()
-
-
-def create_sql_resource(settings: Settings) -> SQLResource:
-    engine = create_engine(
-        url=str(settings.postgres_dsn),
-        **settings.postgres_params,
-    )
-    with engine.connect() as connection:
-        connection.execute(text("SELECT 1"))
-    logger.info("SQL engine up")
-    return SQLResource(
-        engine=engine,
-        session_factory=sessionmaker(bind=engine),
-    )
-
-
-@contextmanager
-def managed_session(session_factory: sessionmaker[Session]) -> Iterator[Session]:
-    session = session_factory()
-    try:
-        yield session
-        session.commit()
-        logger.info("Commit ok")
-    except Exception as error:
-        logger.error(f"Rollback due to {error}")
-        session.rollback()
-        raise
-    finally:
-        session.close()

tests/api/conftest.py

@@ -4,12 +4,13 @@
 import pytest
 from django.conf import settings
 from django.test import Client
+from fast_depends import dependency_provider
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 from flask import Flask
 
+from app.api.dependencies import get_settings
 from app.api.fastapi.app import create_fastapi_app
-from app.api.fastapi.dependencies import get_settings
 from app.api.flask.app import create_flask_app
 from app.core.settings import Settings
 from tests.api.clients.base import HTTPClient
@@ -28,6 +29,7 @@ def fastapi_app(app_settings: Settings) -> FastAPI:
 @pytest.fixture(scope="session")
 def flask_app(app_settings: Settings) -> Flask:
     app = create_flask_app(settings=app_settings)
+    dependency_provider.override(get_settings, settings_override_func)
     return app
 
 
@@ -43,6 +45,7 @@ def django_setup(app_settings: Settings) -> None:
         ALLOWED_HOSTS=["testserver"],
     )
     django.setup()
+    dependency_provider.override(get_settings, settings_override_func)
 
 
 @pytest.fixture

tests/api/items/test_items_special.py

@@ -4,10 +4,20 @@
 from fastapi import status
 from sqlalchemy.orm import Session
 
+from app.core.settings import AppEnvironment
 from app.infrastructure.sqlalchemy.models.items import SQLItemModel
 from tests.api.clients.base import HTTPClient
 
 
+@pytest.mark.parametrize("client", ["fastapi", "flask", "django"], indirect=True)
+def test_get_env(client: HTTPClient) -> None:
+    response = client.get("/dev/env")
+
+    assert response.status_code == status.HTTP_200_OK
+    result = response.json()
+    assert result["environment"] == AppEnvironment.TESTING
+
+
 @pytest.mark.parametrize("client", ["fastapi", "flask", "django"], indirect=True)
 def test_item_domain_error(client: HTTPClient, session: Session) -> None:
     item_id = uuid.uuid7()