Feature/protocol refactor

app/business/todo_management/services/todo.py

@@ -4,15 +4,15 @@
 
 from app.business.todo_management.models.todo import TodoDto, CreateTodoRequest, TodoID
 from app.domain.todo_management.models import Todo
-from app.domain.todo_management.repositories.todo import TodoSQLRepository
+from app.domain.todo_management.repositories.todo import TodoSQLRepository, TodoRepositoryProtocol
 
 
 def parse_to_dto(todo_entity: Todo):
     return TodoDto(**todo_entity.dict())
 
 
 class TodoService:
-    def __init__(self, repository: TodoSQLRepository = Depends(TodoSQLRepository)):
+    def __init__(self, repository: TodoRepositoryProtocol = Depends(TodoSQLRepository)):
         self.todo_repo = repository
 
     async def get_pending_todos(self) -> List[TodoDto]:

app/common/base/base_repository.py

@@ -1,4 +1,4 @@
-from typing import Generic, TypeVar, Type, Union, Optional
+from typing import Generic, TypeVar, Type, Union, Optional, Protocol
 from uuid import UUID
 
 from fastapi import Depends
@@ -11,6 +11,17 @@
 ModelType = TypeVar("ModelType", bound=SQLModel)
 
 
+class AsyncRepositoryProtocol(Protocol):
+    async def get(self, *, uid: Union[UUID, str]) -> Optional[ModelType]:
+        ...
+
+    async def add(self, *, model: ModelType):
+        ...
+
+    async def save(self, *, model: ModelType, refresh=True):
+        ...
+
+
 class BaseSQLRepository(Generic[ModelType]):
 
     def __init__(self, model: Type[ModelType], session: AsyncSession = Depends(get_async_session)):

app/common/controllers/keycloak/auth.py

@@ -1,15 +1,17 @@
 import logging
 
 from fastapi import APIRouter, Depends
-from fastapi_keycloak import OIDCUser, UsernamePassword
+from fastapi_keycloak import OIDCUser, UsernamePassword, FastAPIKeycloak
 from starlette.responses import RedirectResponse
 
-from app.common import get_user, idp
+from app.common import get_user, idp as _idp
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/auth")
 
+idp: FastAPIKeycloak = _idp.client
+
 #################################
 # Basic Authentication Router
 #################################

app/common/controllers/keycloak/identity.py

@@ -1,16 +1,18 @@
 from typing import Optional, List
 
 from fastapi import APIRouter, Depends, Body, Query
-from fastapi_keycloak import HTTPMethod, KeycloakUser
+from fastapi_keycloak import HTTPMethod, KeycloakUser, FastAPIKeycloak
 from pydantic import SecretStr
-from app.common import get_user, idp
+from app.common import get_user, idp as _idp
 
 router = APIRouter(prefix="/idp", dependencies=[Depends(get_user())])  # Protect all the paths with user authentication
 
 #################################
 # IDP Admin Router
 #################################
 
+idp: FastAPIKeycloak = _idp.client
+
 
 @router.post("/proxy", tags=["admin-cli"])
 def proxy_admin_request(relative_path: str, method: HTTPMethod, additional_headers: dict = Body(None),

app/common/core/__init__.py

@@ -29,6 +29,7 @@ def get_api(routers: List[APIRouter]):
     # Configure CORS
     configure_cors(app_settings, api)
     # Init infra
-    from app.common.infra import configure_api
-    configure_api(api)
+    from app.common import idp
+    if idp is not None:
+        idp.configure_api(api)
     return api

app/common/core/identity_provider.py

@@ -1,5 +1,7 @@
 import abc
-from typing import Protocol, Optional, List, Callable, Any
+from typing import Protocol, Optional, List, Callable, Any, T
+
+from fastapi import FastAPI
 from pydantic import BaseModel
 
 
@@ -9,9 +11,8 @@ class User(BaseModel):
     roles: List[str]
 
 
-class IdentityProvider(abc.ABC):
-    @abc.abstractmethod
-    def get_current_user(self, required_roles: list[str] | None = None):
+class IdentityProvider(Protocol):
+    def get_current_user(self, required_roles: list[str] | None = None) -> Callable[[Any], User]:
         """Function that checks the current user based on an access token in the HTTP-header. Optionally verifies
         roles are possessed by the user
 
@@ -27,4 +28,13 @@ def get_current_user(self, required_roles: list[str] | None = None):
             JWTClaimsError: If any claim is invalid
             HTTPException: If any role required is not contained within the roles of the users
         """
-        raise NotImplementedError
+        ...
+
+    @property
+    def client(self) -> T:
+        raise Exception
+
+    def configure_api(self, api: FastAPI):
+        """Function that configures the API adding extra controllers or features specifics for this IDP
+        """
+        ...

app/common/infra/__init__.py

@@ -1,13 +1,12 @@
 import logging
 from enum import Enum
 
-from fastapi import FastAPI
 from pydantic import ValidationError
 
 from app.common.core.configuration import load_env_file_on_settings
-from app.common.infra.firebase import FirebaseSettings, get_firebase_settings, FirebaseService, configure_firebase_api
+from app.common.infra.firebase import FirebaseSettings, get_firebase_settings, FirebaseService
 from app.common.core.identity_provider import IdentityProvider
-from app.common.infra.keycloak import KeycloakSettings, get_keycloak_settings, KeycloakService, configure_keycloak_api
+from app.common.infra.keycloak import KeycloakSettings, get_keycloak_settings, KeycloakService
 
 
 logger = logging.getLogger(__name__)
@@ -32,15 +31,7 @@ def __get_idp() -> (IdentityProvider, IDPType):
     return None, None
 
 
-# TODO: Add this as class method (and refactor this file)
-def configure_api(api: FastAPI):
-    if idp_type == IDPType.KEYCLOAK:
-        configure_keycloak_api(api)
-    if idp_type == IDPType.FIREBASE:
-        configure_firebase_api(api)
-
-
 idp_configuration = __get_idp()
 
-idp: IdentityProvider = idp_configuration[0]
-idp_type: IDPType = idp_configuration[1]
+idp: IdentityProvider | None = idp_configuration[0]
+idp_type: IDPType | None = idp_configuration[1]

app/common/infra/firebase.py

@@ -2,6 +2,7 @@
 from typing import Optional
 from fastapi import Depends, FastAPI
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from firebase_admin import App
 from pydantic import BaseSettings
 
 from app.common.core.configuration import load_env_file_on_settings
@@ -24,14 +25,18 @@ def get_firebase_settings() -> FirebaseSettings:
     return load_env_file_on_settings(FirebaseSettings)
 
 
-class FirebaseService(IdentityProvider):
+class FirebaseService:
     def __init__(self, settings: FirebaseSettings):
         import firebase_admin
         firebase_credentials = firebase_admin.credentials.Certificate(settings.credentials_file)
         options = {}
         if settings.database_url:
             options['databaseURL'] = settings.database_url
-        self.client = firebase_admin.initialize_app(credential=firebase_credentials, options=options)
+        self._client = firebase_admin.initialize_app(credential=firebase_credentials, options=options)
+
+    @property
+    def client(self) -> App:
+        return self._client
 
     def get_current_user(self, required_roles: list[str] | None = None):
         from firebase_admin import auth
@@ -55,10 +60,8 @@ def validate_token(credential: HTTPAuthorizationCredentials = Depends(HTTPBearer
 
         return validate_token
 
-
-def configure_firebase_api(api: FastAPI):
-    from app.common import idp
-    if idp is not None:
+    @staticmethod
+    def configure_api(api: FastAPI):
         # Include auth router
         from app.common.controllers import firebase_routers
         api.include_router(firebase_routers)

app/common/infra/keycloak.py

@@ -1,9 +1,9 @@
 from functools import lru_cache
 
 from fastapi import FastAPI
+from fastapi_keycloak import FastAPIKeycloak
 from pydantic import BaseSettings
 
-from app.common.core.identity_provider import IdentityProvider
 from app.common.core.configuration import load_env_file_on_settings
 
 
@@ -25,26 +25,27 @@ def get_keycloak_settings() -> KeycloakSettings:
     return load_env_file_on_settings(KeycloakSettings)
 
 
-def configure_keycloak_api(api: FastAPI):
-    from app.common import idp
-    if idp is not None:
-        # Enable authentication layer to swagger endpoints
-        idp.add_swagger_config(api)
-        # Include auth router
-        from app.common.controllers import keycloak_routers
-        api.include_router(keycloak_routers)
-
-
-class KeycloakService(IdentityProvider):
+class KeycloakService:
 
     def __init__(self, keycloak_settings: KeycloakSettings):
         from fastapi_keycloak import FastAPIKeycloak
-        self.client = FastAPIKeycloak(server_url=keycloak_settings.auth_server,
-                                      client_id=keycloak_settings.client_id,
-                                      client_secret=keycloak_settings.client_secret,
-                                      admin_client_secret=keycloak_settings.admin_client_secret,
-                                      realm=keycloak_settings.realm,
-                                      callback_uri=keycloak_settings.callback_uri)
+        self._client = FastAPIKeycloak(server_url=keycloak_settings.auth_server,
+                                       client_id=keycloak_settings.client_id,
+                                       client_secret=keycloak_settings.client_secret,
+                                       admin_client_secret=keycloak_settings.admin_client_secret,
+                                       realm=keycloak_settings.realm,
+                                       callback_uri=keycloak_settings.callback_uri)
+
+    @property
+    def client(self) -> FastAPIKeycloak:
+        return self._client
 
     def get_current_user(self, required_roles: list[str] | None = None):
         return self.client.get_current_user(required_roles=required_roles)  # TODO: Refactor to return User
+
+    def configure_api(self, api: FastAPI):
+        # Enable authentication layer to swagger endpoints
+        self.client.add_swagger_config(api)
+        # Include auth router
+        from app.common.controllers import keycloak_routers
+        api.include_router(keycloak_routers)

app/domain/todo_management/repositories/todo.py

@@ -1,15 +1,26 @@
-from typing import List
+from typing import List, Protocol
 from uuid import UUID
 
 from fastapi import Depends
 from sqlmodel.sql.expression import Select, select
 
-from app.common.base.base_repository import BaseSQLRepository
+from app.common.base.base_repository import BaseSQLRepository, AsyncRepositoryProtocol
 from app.common.exceptions.http import NotFoundException
 from app.common.infra.sql_adaptors import get_session, get_async_session, AsyncSession
 from app.domain.todo_management.models import Todo
 
 
+class TodoRepositoryProtocol(Protocol):
+    async def create(self, *, description: str) -> Todo:
+        ...
+
+    async def get_pending_todos(self) -> List[Todo]:
+        ...
+
+    async def todo_done(self, todo_id: UUID) -> Todo:
+        ...
+
+
 class TodoSQLRepository(BaseSQLRepository[Todo]):
 
     def __init__(self, session: AsyncSession = Depends(get_async_session)):