Updated src/main, because on changing the FastAPI version on_event(st… (

#100)

…artup) was depecated in the new version.

Incorporated ragtech_commons_api and deleted the old ones

---------

Co-authored-by: Nargis Sultani <[email protected]>

pyproject.toml

@@ -8,7 +8,7 @@ packages = [{ include = "regtech-user-fi-management" }]
 
 [tool.poetry.dependencies]
 python = "^3.11"
-fastapi = "^0.103.1"
+fastapi = "^0.104.1"
 uvicorn = "^0.22.0"
 python-dotenv = "^1.0.0"
 python-keycloak = "^3.0.0"
@@ -19,6 +19,7 @@ requests = "^2.31.0"
 asyncpg = "^0.27.0"
 alembic = "^1.12.0"
 pydantic-settings = "^2.0.3"
+regtech-api-commons = {git = "https://github.com/cfpb/regtech-api-commons.git"}
 
 [tool.poetry.group.dev.dependencies]
 ruff = "0.0.278"

src/config.py

@@ -1,12 +1,13 @@
 import os
 from urllib import parse
-from typing import Dict, Any
+from typing import Any
 
-from pydantic import TypeAdapter, field_validator, ValidationInfo
-from pydantic.networks import HttpUrl, PostgresDsn
-from pydantic.types import SecretStr
+from pydantic import field_validator, ValidationInfo
+from pydantic.networks import PostgresDsn
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from regtech_api_commons.oauth2.config import KeycloakSettings
+
 
 JWT_OPTS_PREFIX = "jwt_opts_"
 
@@ -23,20 +24,9 @@ class Settings(BaseSettings):
     inst_db_host: str
     inst_db_scheme: str = "postgresql+asyncpg"
     inst_conn: PostgresDsn | None = None
-    auth_client: str
-    auth_url: HttpUrl
-    token_url: HttpUrl
-    certs_url: HttpUrl
-    kc_url: HttpUrl
-    kc_realm: str
-    kc_admin_client_id: str
-    kc_admin_client_secret: SecretStr
-    kc_realm_url: HttpUrl
-    jwt_opts: Dict[str, bool | int] = {}
 
     def __init__(self, **data):
         super().__init__(**data)
-        self.set_jwt_opts()
 
     @field_validator("inst_conn", mode="before")
     @classmethod
@@ -50,31 +40,9 @@ def build_postgres_dsn(cls, postgres_dsn, info: ValidationInfo) -> Any:
         )
         return str(postgres_dsn)
 
-    def set_jwt_opts(self) -> None:
-        """
-        Converts `jwt_opts_` prefixed settings, and env vars into JWT options dictionary.
-        all options are boolean, with exception of 'leeway' being int
-        valid options can be found here:
-        https://github.com/mpdavis/python-jose/blob/4b0701b46a8d00988afcc5168c2b3a1fd60d15d8/jose/jwt.py#L81
-
-        Because we're using model_extra to load in jwt_opts as a dynamic dictionary,
-        normal env overrides does not take place on top of dotenv files,
-        so we're merging settings.model_extra with environment variables.
-        """
-        jwt_opts_adapter = TypeAdapter(int | bool)
-        self.jwt_opts = {
-            **self.parse_jwt_vars(jwt_opts_adapter, self.model_extra.items()),
-            **self.parse_jwt_vars(jwt_opts_adapter, os.environ.items()),
-        }
-
-    def parse_jwt_vars(self, type_adapter: TypeAdapter, setting_variables: Dict[str, Any]) -> Dict[str, bool | int]:
-        return {
-            key.lower().replace(JWT_OPTS_PREFIX, ""): type_adapter.validate_python(value)
-            for (key, value) in setting_variables
-            if key.lower().startswith(JWT_OPTS_PREFIX)
-        }
-
     model_config = SettingsConfigDict(env_file=env_files_to_load, extra="allow")
 
 
 settings = Settings()
+
+kc_settings = KeycloakSettings(_env_file=env_files_to_load)

src/entities/models/__init__.py

@@ -10,7 +10,6 @@
     "DeniedDomainDao",
     "DeniedDomainDto",
     "UserProfile",
-    "AuthenticatedUser",
     "FederalRegulatorDao",
     "HMDAInstitutionTypeDao",
     "SBLInstitutionTypeDao",
@@ -42,7 +41,6 @@
     FinanicialInstitutionAssociationDto,
     DeniedDomainDto,
     UserProfile,
-    AuthenticatedUser,
     FederalRegulatorDto,
     InstitutionTypeDto,
     AddressStateDto,

src/entities/models/dto.py

@@ -1,6 +1,5 @@
-from typing import List, Dict, Any, Set
+from typing import List, Set
 from pydantic import BaseModel, model_validator
-from starlette.authentication import BaseUser
 
 
 class FinancialInsitutionDomainBase(BaseModel):
@@ -132,46 +131,3 @@ class FinancialInstitutionWithRelationsDto(FinancialInstitutionDto):
 
 class FinanicialInstitutionAssociationDto(FinancialInstitutionWithRelationsDto):
     approved: bool
-
-
-class AuthenticatedUser(BaseUser, BaseModel):
-    claims: Dict[str, Any]
-    name: str
-    username: str
-    email: str
-    id: str
-    institutions: List[str]
-
-    @classmethod
-    def from_claim(cls, claims: Dict[str, Any]) -> "AuthenticatedUser":
-        return cls(
-            claims=claims,
-            name=claims.get("name", ""),
-            username=claims.get("preferred_username", ""),
-            email=claims.get("email", ""),
-            id=claims.get("sub", ""),
-            institutions=cls.parse_institutions(claims.get("institutions")),
-        )
-
-    @classmethod
-    def parse_institutions(cls, institutions: List[str] | None) -> List[str]:
-        """
-        Parse out the list of institutions returned by Keycloak
-
-        Args:
-            institutions(List[str]): list of full institution paths provided by keycloak,
-            it is possible to have nested paths, though we may not use the feature.
-            e.g. ["/ROOT_INSTITUTION/CHILD_INSTITUTION/GRAND_CHILD_INSTITUTION"]
-
-        Returns:
-            List[str]: List of cleaned up institutions.
-            e.g. ["GRAND_CHILD_INSTITUTION"]
-        """
-        if institutions:
-            return [institution.split("/")[-1] for institution in institutions]
-        else:
-            return []
-
-    @property
-    def is_authenticated(self) -> bool:
-        return True

src/main.py

@@ -1,3 +1,4 @@
+from contextlib import asynccontextmanager
 import os
 import logging
 from http import HTTPStatus
@@ -11,24 +12,34 @@
 
 from routers import admin_router, institutions_router
 
-from oauth2 import BearerTokenAuthBackend
+from regtech_api_commons.oauth2.oauth2_backend import BearerTokenAuthBackend
+from regtech_api_commons.oauth2.oauth2_admin import OAuth2Admin
 
-from config import settings
+from config import kc_settings
 
 log = logging.getLogger()
 
-app = FastAPI()
 
-
-@app.on_event("startup")
-async def app_start():
+def run_migrations():
     file_dir = os.path.dirname(os.path.realpath(__file__))
     alembic_cfg = Config(f"{file_dir}/../alembic.ini")
     alembic_cfg.set_main_option("script_location", f"{file_dir}/../db_revisions")
     alembic_cfg.set_main_option("prepend_sys_path", f"{file_dir}/../")
     command.upgrade(alembic_cfg, "head")
 
 
+@asynccontextmanager
+async def lifespan(app_: FastAPI):
+    log.info("Starting up...")
+    log.info("run alembic upgrade head...")
+    run_migrations()
+    yield
+    log.info("Shutting down...")
+
+
+app = FastAPI(lifespan=lifespan)
+
+
 @app.exception_handler(HTTPException)
 async def http_exception_handler(request: Request, exception: HTTPException) -> JSONResponse:
     log.error(exception, exc_info=True, stack_info=True)
@@ -45,10 +56,10 @@ async def general_exception_handler(request: Request, exception: Exception) -> J
 
 
 oauth2_scheme = OAuth2AuthorizationCodeBearer(
-    authorizationUrl=settings.auth_url.unicode_string(), tokenUrl=settings.token_url.unicode_string()
+    authorizationUrl=kc_settings.auth_url.unicode_string(), tokenUrl=kc_settings.token_url.unicode_string()
 )
 
-app.add_middleware(AuthenticationMiddleware, backend=BearerTokenAuthBackend(oauth2_scheme))
+app.add_middleware(AuthenticationMiddleware, backend=BearerTokenAuthBackend(oauth2_scheme, OAuth2Admin(kc_settings)))
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],

src/routers/admin.py

@@ -3,14 +3,18 @@
 from fastapi import Depends, Request
 from starlette.authentication import requires
 from dependencies import check_domain
-from util import Router
+
+from regtech_api_commons.api import Router
 from entities.models import UserProfile
 
-from entities.models import AuthenticatedUser
-from oauth2 import oauth2_admin
+from regtech_api_commons.models.auth import AuthenticatedUser
+from regtech_api_commons.oauth2.oauth2_admin import OAuth2Admin
+from config import kc_settings
 
 router = Router()
 
+oauth2_admin = OAuth2Admin(kc_settings)
+
 
 @router.get("/me/", response_model=AuthenticatedUser)
 @requires("authenticated")