refactor: py39 standard collection type hinting

kingdom_sdk/adapters/message_bus.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, AsyncGenerator, Callable, Dict, List, Type
+from typing import Any, AsyncGenerator, Callable, Type
 
 from kingdom_sdk.domain.exception import KingdomError
 from kingdom_sdk.domain.message import Command, Event, Message
@@ -14,16 +14,16 @@
 
 class MessageBus(AbstractMessageBus):
     _uow: AbstractUnitOfWork
-    _event_handlers: Dict[Type[Event], List[Callable]]
-    _command_handlers: Dict[Type[Command], Callable]
-    _queue: List[Message]
+    _event_handlers: dict[Type[Event], list[Callable]]
+    _command_handlers: dict[Type[Command], Callable]
+    _queue: list[Message]
 
     def __init__(
         self,
         uow: AbstractUnitOfWork,
-        event_handlers: Dict[Type[Event], List[Callable]],
-        command_handlers: Dict[Type[Command], Callable],
-        queue: List[Message],
+        event_handlers: dict[Type[Event], list[Callable]],
+        command_handlers: dict[Type[Command], Callable],
+        queue: list[Message],
     ) -> None:
         self._uow = uow
         self._event_handlers = event_handlers
@@ -34,9 +34,9 @@ def __init__(
     def create(
         cls,
         uow: AbstractUnitOfWork,
-        event_handlers: Dict[Type[Event], List[Callable]],
-        command_handlers: Dict[Type[Command], Callable],
-        dependencies: Dict[str, Any],
+        event_handlers: dict[Type[Event], list[Callable]],
+        command_handlers: dict[Type[Command], Callable],
+        dependencies: dict[str, Any],
     ) -> MessageBus:
         """Create a message bus with its handlers dependencies set
         programmatically.
@@ -95,7 +95,7 @@ def _handle_map(self, message: Message) -> Callable:
         else:
             raise UnknownMessage()
 
-    async def handle(self, message: Message) -> List[Warning]:
+    async def handle(self, message: Message) -> list[Warning]:
         self._queue = [message]
         warnings = []
         while self._queue:
@@ -108,7 +108,7 @@ async def handle(self, message: Message) -> List[Warning]:
         return warnings
 
     @staticmethod
-    async def _run(handler: AsyncGenerator) -> List[Any]:
+    async def _run(handler: AsyncGenerator) -> list[Any]:
         return [r async for r in handler]
 
 

kingdom_sdk/adapters/query.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Tuple
+from typing import Any
 
 from jinjasql import JinjaSql
 from sqlalchemy.engine import CursorResult
@@ -15,7 +15,7 @@ class JinjaTemplateSqlMixin(AbstractTemplateSQLMixin):
 
     _sql_file_path: str
 
-    def _build_statement(self, **params: Any) -> Tuple[str, Dict]:
+    def _build_statement(self, **params: Any) -> tuple[str, dict]:
         with open(self._sql_file_path) as sql_file:
             file_content = sql_file.read()
         query, bind_params = self._jinja.prepare_query(file_content, params)

kingdom_sdk/adapters/repository.py

@@ -1,5 +1,5 @@
 from abc import ABC
-from typing import List, Optional, Set, Type
+from typing import List, Optional, Type
 
 from sqlalchemy.orm import Query, Session
 
@@ -20,11 +20,11 @@ class SQLAlchemyRepository(AbstractRepository, ABC):
     __model__: Type[Aggregate]
 
     _session: Session
-    _seen: Set[Aggregate]
+    _seen: set[Aggregate]
 
     def __init__(self, session: Session) -> None:
         self._session = session
-        self._seen: Set[Aggregate] = set()
+        self._seen: set[Aggregate] = set()
 
     @property
     def query(self) -> Query:

kingdom_sdk/adapters/unit_of_work.py

@@ -1,5 +1,5 @@
 from abc import ABC
-from typing import Any, Generator, Iterator, List, Set, Tuple
+from typing import Any, Generator, Iterator
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
@@ -30,7 +30,7 @@ class SQLAlchemyUnitOfWork(AbstractUnitOfWork, ABC):
     ...     repository: ...
     """
 
-    _errors: List[Any]
+    _errors: list[Any]
     _session_factory: sessionmaker
     _session: Session
 
@@ -59,7 +59,7 @@ def execute_native_statement(self, statement: str, **params: Any) -> Any:
         return self._session.execute(statement, params)
 
     def collect_new_events(self) -> Generator:
-        dirty: Set[Aggregate] = set()
+        dirty: set[Aggregate] = set()
 
         for field_name, _ in self._repositories:
             try:
@@ -79,7 +79,7 @@ def _initialize_repositories(self, session: Session) -> None:
             self.__dict__[field_name] = repository(session)
 
     @property
-    def _repositories(self) -> Iterator[Tuple[str, Any]]:
+    def _repositories(self) -> Iterator[tuple[str, Any]]:
         return (
             (field, module)
             for field, module in self.__annotations__.items()

kingdom_sdk/domain/aggregate.py

@@ -1,6 +1,5 @@
 from abc import ABC
 from datetime import datetime
-from typing import List
 from uuid import UUID
 
 from kingdom_sdk.domain.entity import Entity
@@ -10,7 +9,7 @@
 class Aggregate(Entity, ABC):
     """Base class for aggregates."""
 
-    _events: List[Event]
+    _events: list[Event]
 
     def __init__(
         self,
@@ -36,5 +35,5 @@ def next_event(self) -> Event:
         return self._events.pop(0)
 
     @property
-    def events(self) -> List[Event]:
+    def events(self) -> list[Event]:
         return self._events

kingdom_sdk/domain/message.py

@@ -3,7 +3,7 @@
 from abc import ABC
 from dataclasses import asdict, dataclass
 from datetime import datetime
-from typing import Any, Dict
+from typing import Any
 from uuid import UUID
 
 from kingdom_sdk.domain.value_object import ValueObject
@@ -62,7 +62,7 @@ class Event(Message, ABC):
 class PersistentMessage(ValueObject):
     module: str
     classname: str
-    data: Dict[str, Any]
+    data: dict[str, Any]
 
     @classmethod
     def create(cls, message: Message) -> PersistentMessage:  # type: ignore

kingdom_sdk/ports/message_bus.py

@@ -2,7 +2,7 @@
 
 import inspect
 from abc import ABC, abstractmethod
-from typing import Any, Callable, Dict, List, Type
+from typing import Any, Callable, Type
 
 from kingdom_sdk.domain.message import Command, Event, Message
 from kingdom_sdk.ports.unit_of_work import AbstractUnitOfWork
@@ -14,19 +14,19 @@ class AbstractMessageBus(ABC):
     def create(
         cls,
         uow: AbstractUnitOfWork,
-        event_handlers: Dict[Type[Event], List[Callable]],
-        command_handlers: Dict[Type[Command], Callable],
-        dependencies: Dict[str, Any],
+        event_handlers: dict[Type[Event], list[Callable]],
+        command_handlers: dict[Type[Command], Callable],
+        dependencies: dict[str, Any],
     ) -> AbstractMessageBus:
         raise NotImplementedError
 
     @abstractmethod
-    async def handle(self, message: Message) -> List[Warning]:
+    async def handle(self, message: Message) -> list[Warning]:
         raise NotImplementedError
 
     @staticmethod
     def _inject_dependencies(
-        handler: Callable, dependencies: Dict[str, Any]
+        handler: Callable, dependencies: dict[str, Any]
     ) -> Callable:
         """Inspect a handler function to figure out its arguments and returns
         the same handler with its arguments already set given a dependencies

kingdom_sdk/ports/query.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Tuple
+from typing import Any
 
 from kingdom_sdk.ports.unit_of_work import AbstractUnitOfWork
 
@@ -12,7 +12,7 @@ def _build_statement(self) -> str:
 
 class AbstractTemplateSQLMixin(ABC):
     @abstractmethod
-    def _build_statement(self, **params: Any) -> Tuple[str, Dict]:
+    def _build_statement(self, **params: Any) -> tuple[str, dict]:
         raise NotImplementedError
 
 
@@ -26,7 +26,7 @@ class AbstractReadQuery(ABC):
     @abstractmethod
     def execute(
         self, uow: AbstractUnitOfWork, commit: bool = False, **params: Any
-    ) -> Dict:
+    ) -> dict:
         raise NotImplementedError
 
 

kingdom_sdk/ports/repository.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional
+from typing import Optional
 
 from kingdom_sdk.database.types import PrimaryKey_T
 from kingdom_sdk.domain.aggregate import Aggregate
@@ -11,7 +11,7 @@ def add(self, aggregate: Aggregate) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def list(self) -> List[Aggregate]:
+    def list(self) -> list[Aggregate]:
         raise NotImplementedError
 
     @abstractmethod

kingdom_sdk/utils/casting.py

@@ -1,6 +1,3 @@
-from typing import Tuple
-
-
 def bool_from_string(string: str) -> bool:
     string = string.strip().lower()
     if string in ("true", "yes", "t", "y", "1"):
@@ -11,6 +8,6 @@ def bool_from_string(string: str) -> bool:
         raise ValueError("String doesn't represent a valid boolean")
 
 
-def split_module_class(module: str) -> Tuple[str, str]:
+def split_module_class(module: str) -> tuple[str, str]:
     split = module.split(".")
     return ".".join(split[:-1]), split[-1]

kingdom_sdk/utils/files.py

@@ -1,10 +1,10 @@
 import os
-from typing import Iterable, List
+from typing import Iterable
 
 
 def find(
     file_name: str, base_dir: str, excluded_dirs: Iterable[str] = ()
-) -> List[str]:
+) -> list[str]:
     """Return a list containing the paths of the files found."""
     return [
         os.path.join(root, file_name)