G2P-5620 Refactor database session management to use get_async_session_maker across services

.github/workflows/build-publish.yml

@@ -35,7 +35,7 @@ jobs:
               {
                 "arg": "FASTAPI_COMMON_REF",
                 "repo": "https://github.com/openg2p/openg2p-fastapi-common",
-                "ref": "1.1"
+                "ref": "develop"
               }
             ]
           },
@@ -47,7 +47,7 @@ jobs:
               {
                 "arg": "FASTAPI_COMMON_REF",
                 "repo": "https://github.com/openg2p/openg2p-fastapi-common",
-                "ref": "1.1"
+                "ref": "develop"
               }
             ]
           },
@@ -59,7 +59,7 @@ jobs:
               {
                 "arg": "FASTAPI_COMMON_REF",
                 "repo": "https://github.com/openg2p/openg2p-fastapi-common",
-                "ref": "1.1"
+                "ref": "develop"
               }
             ]
           }

.github/workflows/iam-core-test.yml

@@ -19,7 +19,7 @@ on:
     inputs:
       fastapi_common_ref:
         description: "openg2p-fastapi-common git ref (tag/branch)"
-        default: "1.1"
+        default: "develop"
         type: string
 
 concurrency:
@@ -50,7 +50,7 @@ jobs:
           python -m pip install --upgrade pip
           # External lib (openg2p-fastapi-common) from git.
           python -m pip install \
-            "git+https://github.com/openg2p/openg2p-fastapi-common@${{ inputs.fastapi_common_ref || '1.1' }}#subdirectory=openg2p-fastapi-common"
+            "git+https://github.com/openg2p/openg2p-fastapi-common@${{ inputs.fastapi_common_ref || 'develop' }}#subdirectory=openg2p-fastapi-common"
           # In-repo library (not on PyPI) + test tooling.
           python -m pip install -e ./iam-core
           python -m pip install pytest pytest-asyncio

.github/workflows/iam-staff-portal-api-test.yml

@@ -19,7 +19,7 @@ on:
     inputs:
       fastapi_common_ref:
         description: "openg2p-fastapi-common git ref (tag/branch)"
-        default: "1.1"
+        default: "develop"
         type: string
 
 concurrency:
@@ -49,7 +49,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           python -m pip install \
-            "git+https://github.com/openg2p/openg2p-fastapi-common@${{ inputs.fastapi_common_ref || '1.1' }}#subdirectory=openg2p-fastapi-common"
+            "git+https://github.com/openg2p/openg2p-fastapi-common@${{ inputs.fastapi_common_ref || 'develop' }}#subdirectory=openg2p-fastapi-common"
           python -m pip install -e "./iam-core[dev]"
           python -m pip install -e "./iam-staff-portal-api[dev]"
       - name: Run test suite

deployments/charts/openg2p-iam-service/values.yaml

@@ -143,6 +143,11 @@ iamStaffPortalApi:
     IAM_STAFF_DB_DBNAME: '{{ tpl .Values.global.iamDB $ }}'
     IAM_STAFF_DB_PORT: '{{ .Values.global.iamDBPort }}'
     IAM_STAFF_DB_USERNAME: '{{ tpl .Values.global.iamDBUser $ }}'
+    # SQLAlchemy pool per gunicorn worker process.
+    IAM_STAFF_DB_POOL_SIZE: 5
+    IAM_STAFF_DB_POOL_MAX_OVERFLOW: 10
+    IAM_STAFF_DB_POOL_PRE_PING: true
+    IAM_STAFF_DB_POOL_RECYCLE: 1800
     IAM_STAFF_OPENAPI_ROOT_PATH: '{{ .Values.openapiRootPath }}'
     IAM_STAFF_KEYCLOAK_BASE_URL: '{{ .Values.global.keycloakBaseUrl }}'
     IAM_STAFF_KEYCLOAK_ISSUER_URL: '{{ tpl .Values.global.keycloakExternalIssuerUrl $ }}'

iam-core/pyproject.toml

@@ -15,7 +15,6 @@ classifiers = [
   "Operating System :: OS Independent",
 ]
 dependencies = [
-  "openg2p-fastapi-common",
   "cryptography >=41.0.4",
   "python-jose >=3.3.0",
   "httpx",

iam-core/src/iam_core/context.py

@@ -1,9 +1,6 @@
 from contextvars import ContextVar
 from typing import Any, Dict, List, Optional
 
-jwks_cache: ContextVar[dict] = ContextVar("jwks_cache", default=None)
-server_metadata_cache: ContextVar[dict] = ContextVar("server_metadata_cache", default=None)
-
 auth_id_type_config_cache: ContextVar[Optional[Dict[str, Any]]] = ContextVar(
     "auth_id_type_config_cache", default=None
 )

iam-core/src/iam_core/user_auth/app.py

@@ -16,12 +16,15 @@
 from iam_core.user_auth.adapters.implementations.esignet_adapter import EsignetAdapter
 from iam_core.user_auth.adapters.adapter_factory import AdapterFactory
 from iam_core.partner_auth.jwt_validation_helper import JWTValidationHelper
+from .cache import init_cache
 
 
 class Initializer(BaseInitializer):
     def initialize(self, **kwargs):
         super().initialize()
 
+        init_cache()
+
         # Adapters
         OIDCBase()
         KeycloakAdapter()

iam-core/src/iam_core/user_auth/cache.py

@@ -0,0 +1,6 @@
+from fastapi_cache import FastAPICache
+from fastapi_cache.backends.inmemory import InMemoryBackend
+
+
+def init_cache() -> None:
+    FastAPICache.init(InMemoryBackend(), prefix="iam-cache")

iam-core/src/iam_core/user_auth/config.py

@@ -51,3 +51,8 @@ class Settings(BaseSettings):
     # Client-side cache for role→permissions resolution (ResolvePermissionMiddleware).
     # Same roles share one entry across users; short TTL so IAM role mapping changes propagate.
     auth_permissions_cache_ttl_seconds: int = 60
+
+    # Process-wide caches for IdP discovery (OidcClient / JWKS).
+    # ContextVar cannot be used here — it is request-scoped and would refetch every call.
+    auth_oidc_metadata_cache_ttl_seconds: int = 60 * 5
+    auth_jwks_cache_ttl_seconds: int = 60 * 5

iam-core/src/iam_core/user_auth/helpers/jwks_helper.py

@@ -1,21 +1,36 @@
 import logging
 
 import httpx
+from fastapi_cache.coder import PickleCoder
+from fastapi_cache.decorator import cache
 from openg2p_fastapi_common.errors.http_exceptions import InternalServerError
 
-from iam_core.context import jwks_cache
 from iam_core.user_auth.config import Settings
 
 _config = Settings.get_config(strict=False)
 _logger = logging.getLogger(_config.logging_default_logger_name)
 
 
+def _jwks_key_builder(func, namespace: str, *args, **kwargs) -> str:
+    """Cache by issuer (same as the old ContextVar), falling back to jwks_uri."""
+    call_args = kwargs.get("args") or ()
+    call_kwargs = kwargs.get("kwargs") or {}
+    metadata = call_args[0] if call_args else call_kwargs.get("metadata") or {}
+    if len(call_args) > 1:
+        issuer = call_args[1]
+    else:
+        issuer = call_kwargs.get("issuer")
+    jwks_uri = metadata.get("jwks_uri") if isinstance(metadata, dict) else None
+    return f"{namespace}:jwks:{issuer or jwks_uri or 'missing'}"
+
+
+@cache(
+    expire=_config.auth_jwks_cache_ttl_seconds,
+    key_builder=_jwks_key_builder,
+    coder=PickleCoder,
+)
 async def get_jwks(metadata: dict, issuer: str | None = None) -> dict:
-    """Fetch and cache JWKS for the given OIDC metadata. Uses issuer for cache key."""
-    cache = jwks_cache.get() or {}
-    if issuer and issuer in cache:
-        return cache[issuer]
-
+    """Fetch JWKS for the given OIDC metadata. Cached process-wide by issuer."""
     jwks_url = metadata.get("jwks_uri")
     if not jwks_url and issuer:
         jwks_url = f"{issuer.rstrip('/')}/.well-known/jwks.json"
@@ -28,9 +43,4 @@ async def get_jwks(metadata: dict, issuer: str | None = None) -> dict:
     async with httpx.AsyncClient(verify=_config.auth_verify_ssl, timeout=10) as client:
         response = await client.get(jwks_url)
         response.raise_for_status()
-        jwks = response.json()
-
-    if issuer:
-        cache[issuer] = jwks
-        jwks_cache.set(cache)
-    return jwks
+        return response.json()

iam-core/src/iam_core/user_auth/oidc_client.py

@@ -5,10 +5,11 @@
 import httpx
 from authlib.integrations.httpx_client import AsyncOAuth2Client
 from authlib.oauth2.rfc7636 import create_s256_code_challenge
+from fastapi_cache.coder import PickleCoder
+from fastapi_cache.decorator import cache
 from jose import jwt as jose_jwt
 from openg2p_fastapi_common.errors.http_exceptions import InternalServerError, UnauthorizedError
 
-from iam_core.context import server_metadata_cache
 from iam_core.models import LoginProvider
 from iam_core.schemas import TokenEndpointAuthMethod
 
@@ -26,6 +27,15 @@
 _logger = logging.getLogger(_config.logging_default_logger_name)
 
 
+def _server_metadata_key_builder(func, namespace: str, *args, **kwargs) -> str:
+    """Cache by login-provider id so the ORM object itself is not part of the key."""
+    call_args = kwargs.get("args") or ()
+    call_kwargs = kwargs.get("kwargs") or {}
+    login_provider = call_args[1] if len(call_args) > 1 else call_kwargs.get("login_provider")
+    provider_id = getattr(login_provider, "id", None)
+    return f"{namespace}:oidc_server_metadata:{provider_id}"
+
+
 class OidcClient:
     @staticmethod
     def _extra_params(login_provider: LoginProvider) -> dict:
@@ -71,12 +81,12 @@ def _metadata_url(cls, login_provider: LoginProvider) -> str | None:
             return None
         return f"{issuer.rstrip('/')}/.well-known/openid-configuration"
 
+    @cache(
+        expire=_config.auth_oidc_metadata_cache_ttl_seconds,
+        key_builder=_server_metadata_key_builder,
+        coder=PickleCoder,
+    )
     async def get_server_metadata(self, login_provider: LoginProvider) -> dict:
-        cache = server_metadata_cache.get() or {}
-        cache_key = f"lp:{login_provider.id}"
-        if cache_key in cache:
-            return cache[cache_key]
-
         metadata_url = self._metadata_url(login_provider)
         metadata = {}
         if metadata_url:
@@ -94,8 +104,6 @@ async def get_server_metadata(self, login_provider: LoginProvider) -> dict:
         if login_provider.jwks_uri:
             metadata["jwks_uri"] = login_provider.jwks_uri
 
-        cache[cache_key] = metadata
-        server_metadata_cache.set(cache)
         return metadata
 
     async def build_authorize_redirect(

iam-core/tests/test_helpers_and_middleware.py

@@ -6,6 +6,8 @@
 import pytest
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
+from fastapi_cache import FastAPICache
+from fastapi_cache.backends.inmemory import InMemoryBackend
 from jose import jwt as jose_jwt
 from openg2p_fastapi_common.errors.http_exceptions import (
     ForbiddenError,
@@ -14,7 +16,6 @@
 )
 from starlette.routing import Match
 
-from iam_core.context import jwks_cache
 from iam_core.schemas import TokenEndpointAuthMethod
 from iam_core.user_auth.decorators import (
     endpoint_requires_auth,
@@ -235,7 +236,8 @@ def test_logout_token_helper_validation_paths():
 
 @pytest.mark.asyncio
 async def test_jwks_helper_fetch_and_cache():
-    jwks_cache.set(None)
+    FastAPICache.init(InMemoryBackend(), prefix="test-jwks")
+    await FastAPICache.clear()
     metadata = {"jwks_uri": "https://issuer/jwks"}
     response = MagicMock()
     response.raise_for_status = MagicMock()

iam-core/tests/test_oidc_and_adapters.py

@@ -4,10 +4,11 @@
 import pytest
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
+from fastapi_cache import FastAPICache
+from fastapi_cache.backends.inmemory import InMemoryBackend
 from jose import jwt as jose_jwt
 from openg2p_fastapi_common.errors.http_exceptions import InternalServerError, UnauthorizedError
 
-from iam_core.context import server_metadata_cache
 from iam_core.schemas import TokenEndpointAuthMethod
 from iam_core.user_auth.adapters.adapter_factory import AdapterFactory
 from iam_core.user_auth.adapters.implementations.esignet_adapter import EsignetAdapter
@@ -185,7 +186,8 @@ def test_oidc_client_guess_issuer_from_endpoints():
 
 @pytest.mark.asyncio
 async def test_oidc_client_get_server_metadata_fetches_and_caches():
-    server_metadata_cache.set(None)
+    FastAPICache.init(InMemoryBackend(), prefix="test-oidc-metadata")
+    await FastAPICache.clear()
     lp = make_login_provider(server_metadata_url="https://idp/.well-known/openid-configuration")
     client = OidcClient()
     metadata_response = MagicMock()

iam-core/tests/test_user_auth_app.py

@@ -18,6 +18,7 @@ def test_initializer_registers_auth_components():
         patch("iam_core.user_auth.app.RedisRefreshTokenStore") as redis_refresh_token_store,
         patch("iam_core.user_auth.app.TokenValidatorService") as token_validator_service,
         patch("iam_core.user_auth.app.JWTValidationHelper") as jwt_validation_helper,
+        patch("iam_core.user_auth.app.init_cache") as init_cache,
     ):
         Initializer.initialize(init)
 
@@ -31,3 +32,4 @@ def test_initializer_registers_auth_components():
     redis_refresh_token_store.assert_called_once()
     token_validator_service.assert_called_once()
     jwt_validation_helper.assert_called_once()
+    init_cache.assert_called_once()

iam-staff-portal-api/.env.example

@@ -9,6 +9,11 @@ IAM_STAFF_DB_USERNAME=postgres
 IAM_STAFF_DB_PASSWORD=password
 IAM_STAFF_DB_DBNAME=iam_staff
 IAM_STAFF_DB_PORT=5432
+# SQLAlchemy pool per process. Defaults are 5 / 10.
+IAM_STAFF_DB_POOL_SIZE=5
+IAM_STAFF_DB_POOL_MAX_OVERFLOW=10
+IAM_STAFF_DB_POOL_PRE_PING=true
+IAM_STAFF_DB_POOL_RECYCLE=1800
 
 IAM_STAFF_OPENAPI_ROOT_PATH=/
 

iam-staff-portal-api/pyproject.toml

@@ -15,7 +15,6 @@ classifiers = [
   "Operating System :: OS Independent",
 ]
 dependencies = [
-  "openg2p-fastapi-common",
   "iam-core",
   "fastapi-cache2",
   "python-slugify>=8.0.0",

iam-staff-portal-api/src/iam_staff_portal_api/controllers/data_policy_controller.py

@@ -4,9 +4,7 @@
 
 from fastapi import Request
 from fastapi_cache.decorator import cache
-from sqlalchemy.ext.asyncio import async_sessionmaker
-
-from openg2p_fastapi_common.context import dbengine
+from openg2p_fastapi_common.context import get_async_session_maker
 from ..config import Settings
 from iam_core.schemas.data_policy import (
     AddPolicyRequest,
@@ -88,7 +86,7 @@ async def get_policy(self, request: GetPolicyRequest) -> GetPolicyResponse:
             payload = request.request_body.request_payload
             _logger.info("Getting data policy policy_id=%s", payload.policy_id)
 
-            session_maker = async_sessionmaker(dbengine.get(), expire_on_commit=False)
+            session_maker = get_async_session_maker()
             async with session_maker() as session:
                 policy = await self._service.get_policy(
                     session,
@@ -115,7 +113,7 @@ async def get_all_policies(self, request: GetAllPoliciesRequest) -> GetAllPolici
                 payload.register_id,
             )
 
-            session_maker = async_sessionmaker(dbengine.get(), expire_on_commit=False)
+            session_maker = get_async_session_maker()
             async with session_maker() as session:
                 policies, total = await self._service.get_all_policies(
                     session,
@@ -156,7 +154,7 @@ async def add_policy(self, request: AddPolicyRequest, http_request: Request) ->
             if payload.application_id is not None and not mnemonic.startswith("DP_"):
                 mnemonic = f"DP_{mnemonic}"
 
-            session_maker = async_sessionmaker(dbengine.get(), expire_on_commit=False)
+            session_maker = get_async_session_maker()
             async with session_maker() as session:
                 policy = await self._service.add_policy(
                     policy_mnemonic=mnemonic,
@@ -213,7 +211,7 @@ async def remove_policy(
             payload = request.request_body.request_payload
             _logger.info("Removing data policy policy_id=%s", payload.policy_id)
 
-            session_maker = async_sessionmaker(dbengine.get(), expire_on_commit=False)
+            session_maker = get_async_session_maker()
             async with session_maker() as session:
                 # Get policy first to check if it's an IAM application policy
                 from iam_core.models import DataPolicy
@@ -283,7 +281,7 @@ async def _evaluate_expression_cached(
         policy_mnemonics: list[str],
     ) -> list[DataPolicyData]:
         """Internal cached method for expression evaluation."""
-        session_maker = async_sessionmaker(dbengine.get(), expire_on_commit=False)
+        session_maker = get_async_session_maker()
         async with session_maker() as session:
             policies = await self._service.get_policies_by_mnemonics(
                 policy_mnemonics=policy_mnemonics,