format

app.py

@@ -19,17 +19,23 @@
 from system_notification.infra.http.controller.send_notification_controller import (
     SendNotificationController,
 )
-from system_notification.infra.http.server.fastapi_http_server import FastApiHttpServer
-from system_notification.infra.http.server.flask_http_server import FlaskHttpServer
+from system_notification.infra.http.server.fastapi_http_server import (
+    FastApiHttpServer,
+)
+from system_notification.infra.http.server.flask_http_server import (
+    FlaskHttpServer,
+)
 from tests.infra.http.controller.api_notification_serializer import (
     ApiNotificationSerializer,
 )
 
-server = FastApiHttpServer(app=FastAPI(root_path=f"/{os.environ.get('STAGE', '')}"))
+server = FastApiHttpServer(
+    app=FastAPI(root_path=f"/{os.environ.get('STAGE', '')}")
+)
 # server = FlaskHttpServer()
 factory_caller = NotificationFactoryCaller()
 factory_caller.add_factory(
-    SlackNotificationFactory(slack_token=SETTINGS.get("SLACK_API_TOKEN", ""))
+    SlackNotificationFactory(slack_token=SETTINGS.get('SLACK_API_TOKEN', ''))
 )
 uc = SendNotificationUseCase(factory_caller=factory_caller)
 send_notification_controller = SendNotificationController(
@@ -41,5 +47,5 @@
     http_server=server,
 )
 handler = Mangum(server._app)
-if __name__ == "__main__":
+if __name__ == '__main__':
     server.serve(port=5000)

conftest.py

@@ -1,6 +1,6 @@
 def pytest_itemcollected(item):
     node = item.obj
     if node.__doc__:
-        test_title_parts = item._nodeid.split("::")
+        test_title_parts = item._nodeid.split('::')
         test_title_parts[-1] = node.__doc__.strip()
-        item._nodeid = "::".join(test_title_parts)
+        item._nodeid = '::'.join(test_title_parts)

system_notification/application/send_notification_usecase/send_notification_usecase.py

@@ -5,10 +5,15 @@
     NotificationTarget,
 )
 from system_notification.domain.protocols import NotificationSender
-from system_notification.domain.protocols.factory_caller_protocol import FactoryCaller
-from system_notification.domain.protocols.notification_protocol import Notification
-from system_notification.domain.exceptions.notification_error import TargetNotFound
-
+from system_notification.domain.protocols.factory_caller_protocol import (
+    FactoryCaller,
+)
+from system_notification.domain.protocols.notification_protocol import (
+    Notification,
+)
+from system_notification.domain.exceptions.notification_error import (
+    TargetNotFound,
+)
 
 
 @dataclass
@@ -18,7 +23,7 @@ class SendNotificationInput:
     target: List[NotificationTarget] = field(default_factory=list)
     priority: Literal[0, 1, 2, 3] = 0
     placeholders: Dict[str, str] = field(default_factory=dict)
-    icon: str = ""
+    icon: str = ''
 
 
 @dataclass
@@ -37,10 +42,7 @@ async def execute(
         output: List[SendNotificationOutput] = []
         if not input.target:
             raise TargetNotFound(
-                {
-                    "is_sent": False,
-                    "detail": "missing at least one target"
-                }
+                {'is_sent': False, 'detail': 'missing at least one target'}
             )
         for target in input.target:
             sender: Optional[
@@ -59,6 +61,8 @@ async def execute(
             notification.icon = input.icon
             await sender.send(notification)
             output.append(
-                SendNotificationOutput(status=notification.status, target=target)
+                SendNotificationOutput(
+                    status=notification.status, target=target
+                )
             )
         return output

system_notification/config/config.py

@@ -3,11 +3,11 @@
 from dynaconf import Dynaconf
 
 settings = Dynaconf(
-    settings_files=["settings.toml", ".secrets.toml"],
+    settings_files=['settings.toml', '.secrets.toml'],
     envvar_prefix=False,
 )
-ENVIRON_TYPE = settings.get("environ_type", "dev")
-SETTINGS = settings.get(ENVIRON_TYPE) or defaultdict(lambda: "missing data")
+ENVIRON_TYPE = settings.get('environ_type', 'dev')
+SETTINGS = settings.get(ENVIRON_TYPE) or defaultdict(lambda: 'missing data')
 
 
 # `envvar_prefix` = export envvars with `export DYNACONF_FOO=bar`.

system_notification/domain/decorators/jwt_auth_controller_decorator.py

@@ -2,9 +2,15 @@
 from typing import Any, Callable
 
 from system_notification.domain.protocols.controller_protocol import Controller
-from system_notification.domain.protocols.jwt_adapter_protocol import JWTAdapter
-from system_notification.infra.http.server.helpers.http_request import HttpRequest
-from system_notification.infra.http.server.helpers.http_response import HttpResponse
+from system_notification.domain.protocols.jwt_adapter_protocol import (
+    JWTAdapter,
+)
+from system_notification.infra.http.server.helpers.http_request import (
+    HttpRequest,
+)
+from system_notification.infra.http.server.helpers.http_response import (
+    HttpResponse,
+)
 
 
 class JWTAuthControllerDecorator:
@@ -20,16 +26,20 @@ def __call__(self, controller: Callable, *args, **kwargs) -> Callable:
         self._undecorated_callable = controller
 
         @functools.wraps(controller)
-        async def wrapper(controller_instance: Callable, request: HttpRequest) -> Any:
-            authorization_list = request.headers.get("authorization", "").split(" ")
+        async def wrapper(
+            controller_instance: Callable, request: HttpRequest
+        ) -> Any:
+            authorization_list = request.headers.get(
+                'authorization', ''
+            ).split(' ')
             if len(authorization_list) <= 1 or not self._jwt_adapter.is_valid(
                 data=authorization_list[1]
             ):
                 return HttpResponse(
                     status_code=401,
                     body={
-                        "message": "Not Authorized",
-                        "detail": "invalid API token",
+                        'message': 'Not Authorized',
+                        'detail': 'invalid API token',
                     },
                 )
             self._is_authenticated = True

system_notification/domain/notification_factory_caller.py

@@ -1,15 +1,21 @@
 import contextlib
 from typing import Dict, Literal, Optional
 
-from system_notification.domain.exceptions.notification_error import TargetNotFound
+from system_notification.domain.exceptions.notification_error import (
+    TargetNotFound,
+)
 from system_notification.domain.notifications.notification_target import (
     NotificationTarget,
 )
 from system_notification.domain.protocols.notification_factory_protocol import (
     NotificationFactory,
 )
-from system_notification.domain.protocols.notification_protocol import Notification
-from system_notification.domain.protocols.notification_sender import NotificationSender
+from system_notification.domain.protocols.notification_protocol import (
+    Notification,
+)
+from system_notification.domain.protocols.notification_sender import (
+    NotificationSender,
+)
 
 
 class NotificationFactoryCaller:
@@ -19,18 +25,20 @@ def __init__(self) -> None:
     def add_factory(self, factory: NotificationFactory) -> None:
         self._factories[factory.target_type] = factory
 
-    async def get_sender(self, target: NotificationTarget) -> Optional[NotificationSender]:
+    async def get_sender(
+        self, target: NotificationTarget
+    ) -> Optional[NotificationSender]:
         with contextlib.suppress(KeyError):
             factory = self._factories[target.type]
             return factory.make_sender()
         raise TargetNotFound(
             {
-                "is_sent": False,
-                "destin": {
-                    "target_type": target.type,
-                    "target": target.target,
+                'is_sent': False,
+                'destin': {
+                    'target_type': target.type,
+                    'target': target.target,
                 },
-                "detail": "unknown_handler",
+                'detail': 'unknown_handler',
             }
         )
 

system_notification/domain/notifications/base_notification.py

@@ -7,21 +7,21 @@
     NotificationTarget,
 )
 
-T = TypeVar("T", covariant=True)
+T = TypeVar('T', covariant=True)
 
 
 @dataclass
 class BaseNotification:
     title: str
     content: str
     priority: Literal[0, 1, 2, 3] = 0
-    icon: str = ""
+    icon: str = ''
 
     def __post_init__(self) -> None:
         self._is_sent: bool = False
         self._vars: Dict[str, str] = {}
         self._target: Optional[NotificationTarget] = None
-        self._status: str = "filled"
+        self._status: str = 'filled'
         if not self.priority or type(self.priority) != int:
             self.priority = 0
 
@@ -46,7 +46,7 @@ def status(self) -> str:
 
     @status.setter
     def status(self, status: str) -> None:
-        if status.lower() in ("queued", "sent"):
+        if status.lower() in ('queued', 'sent'):
             self._is_sent = True
         self._status = status
 

system_notification/domain/notifications/factories/slack_notification_factory.py

@@ -1,23 +1,31 @@
 from typing import Literal
 
-from system_notification.domain.notifications.base_notification import BaseNotification
-from system_notification.domain.protocols.notification_protocol import Notification
-from system_notification.domain.protocols.notification_sender import NotificationSender
+from system_notification.domain.notifications.base_notification import (
+    BaseNotification,
+)
+from system_notification.domain.protocols.notification_protocol import (
+    Notification,
+)
+from system_notification.domain.protocols.notification_sender import (
+    NotificationSender,
+)
 from system_notification.infra.notification_handlers.slack_notification_handler import (
     SlackNotificationHandler,
 )
 
 
 class SlackNotificationFactory:
-    target_type = "slack_channel"
+    target_type = 'slack_channel'
 
     def __init__(self, slack_token: str) -> None:
         self._slack_token = slack_token
 
     def make_notificaton(
         self, title: str, content: str, priority: Literal[0, 1, 2, 3] = 0
     ) -> Notification:
-        return BaseNotification(title=title, content=content, priority=priority)
+        return BaseNotification(
+            title=title, content=content, priority=priority
+        )
 
     def make_sender(self) -> NotificationSender:
         return SlackNotificationHandler(token=self._slack_token)

system_notification/domain/notifications/notification_target.py

@@ -15,4 +15,4 @@ def target(self) -> str:
         return str(self._target).lower()
 
     def __str__(self) -> str:
-        return f"{self.type}:{self.target}"
+        return f'{self.type}:{self.target}'

system_notification/domain/protocols/controller_protocol.py

@@ -1,14 +1,18 @@
 from typing import Protocol
 
-from system_notification.infra.http.server.helpers.http_request import HttpRequest
-from system_notification.infra.http.server.helpers.http_response import HttpResponse
+from system_notification.infra.http.server.helpers.http_request import (
+    HttpRequest,
+)
+from system_notification.infra.http.server.helpers.http_response import (
+    HttpResponse,
+)
 
 
 class HttpServer(Protocol):
     def serve(self, port: int = 8000) -> None:
         ...
 
-    def on(self, method: str, url: str, controller: "Controller") -> None:
+    def on(self, method: str, url: str, controller: 'Controller') -> None:
         ...
 
 

system_notification/domain/protocols/factory_caller_protocol.py

@@ -6,15 +6,21 @@
 from system_notification.domain.protocols.notification_factory_protocol import (
     NotificationFactory,
 )
-from system_notification.domain.protocols.notification_protocol import Notification
-from system_notification.domain.protocols.notification_sender import NotificationSender
+from system_notification.domain.protocols.notification_protocol import (
+    Notification,
+)
+from system_notification.domain.protocols.notification_sender import (
+    NotificationSender,
+)
 
 
 class FactoryCaller(Protocol):
     def add_factory(self, factory: NotificationFactory) -> None:
         pass
 
-    async def get_sender(self, target: NotificationTarget) -> Optional[NotificationSender]:
+    async def get_sender(
+        self, target: NotificationTarget
+    ) -> Optional[NotificationSender]:
         pass
 
     async def get_notification(

system_notification/domain/protocols/jwt_adapter_protocol.py

@@ -1,6 +1,6 @@
 from typing import Dict, Protocol, TypeVar
 
-JWT_TYPES = TypeVar("JWT_TYPES", str, bytes, int, float)
+JWT_TYPES = TypeVar('JWT_TYPES', str, bytes, int, float)
 
 
 class JWTAdapter(Protocol):

system_notification/domain/protocols/notification_factory_protocol.py

@@ -1,7 +1,11 @@
 from typing import Literal, Protocol, Tuple
 
-from system_notification.domain.protocols.notification_protocol import Notification
-from system_notification.domain.protocols.notification_sender import NotificationSender
+from system_notification.domain.protocols.notification_protocol import (
+    Notification,
+)
+from system_notification.domain.protocols.notification_sender import (
+    NotificationSender,
+)
 
 
 class NotificationFactory(Protocol):

system_notification/domain/protocols/notification_sender.py

@@ -1,6 +1,8 @@
 from typing import Protocol, runtime_checkable
 
-from system_notification.domain.protocols.notification_protocol import Notification
+from system_notification.domain.protocols.notification_protocol import (
+    Notification,
+)
 
 
 @runtime_checkable