Small updates

Author: ivan-borovets

config/toml_config_manager.py

@@ -38,19 +38,26 @@ def validate_logging_level(*, level: str) -> LoggingLevel:
         raise ValueError(f"Invalid log level: '{level}'.") from err
 
 
-def configure_logging(*, level: LoggingLevel = DEFAULT_LOG_LEVEL) -> None:
-    logging.getLogger().handlers.clear()
-
+FMT: Final[str] = (
+    "[%(asctime)s.%(msecs)03d] "
+    "[%(threadName)s] "
+    "%(funcName)20s "
+    "%(module)s:%(lineno)d "
+    "%(levelname)-8s - "
+    "%(message)s"
+)
+DATEFMT: Final[str] = "%Y-%m-%d %H:%M:%S"
+
+
+def configure_logging(
+    *,
+    level: LoggingLevel = DEFAULT_LOG_LEVEL,
+) -> None:
     logging.basicConfig(
-        level=getattr(logging, level),
-        datefmt="%Y-%m-%d %H:%M:%S",
-        format=(
-            "[%(asctime)s.%(msecs)03d] "
-            "%(funcName)20s "
-            "%(module)s:%(lineno)d "
-            "%(levelname)-8s - "
-            "%(message)s"
-        ),
+        level=level,
+        datefmt=DATEFMT,
+        format=FMT,
+        force=True,
     )
 
 

src/app/application/common/ports/user_query_gateway.py

@@ -1,14 +1,32 @@
 from abc import abstractmethod
-from typing import Protocol
+from typing import Protocol, TypedDict
+from uuid import UUID
 
-from app.application.common.query_models.user import UserQueryModel
-from app.application.common.query_params.user import UserListParams
+from app.application.common.query_params.offset_pagination import OffsetPaginationParams
+from app.application.common.query_params.sorting import SortingParams
+from app.domain.enums.user_role import UserRole
+
+
+class UserQueryModel(TypedDict):
+    id_: UUID
+    username: str
+    role: UserRole
+    is_active: bool
+
+
+class ListUsersQM(TypedDict):
+    users: list[UserQueryModel]
+    total: int
 
 
 class UserQueryGateway(Protocol):
     @abstractmethod
     async def read_all(
         self,
-        user_read_all_params: UserListParams,
-    ) -> list[UserQueryModel] | None:
-        """:raises ReaderError:"""
+        pagination: OffsetPaginationParams,
+        sorting: SortingParams,
+    ) -> ListUsersQM:
+        """
+        :raises SortingError:
+        :raises ReaderError:
+        """

src/app/application/common/query_models/__init__.py

@@ -4,7 +4,7 @@
 
 
 @dataclass(frozen=True, slots=True, kw_only=True)
-class Pagination:
+class OffsetPaginationParams:
     """
     raises PaginationError
     """

src/app/application/common/query_models/user.py

@@ -1,6 +1,13 @@
+from dataclasses import dataclass
 from enum import StrEnum
 
 
 class SortingOrder(StrEnum):
     ASC = "ASC"
     DESC = "DESC"
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class SortingParams:
+    field: str
+    order: SortingOrder

…cation/common/query_params/pagination.py …common/query_params/offset_pagination.pysrc/app/application/common/query_params/pagination.py renamed to src/app/application/common/query_params/offset_pagination.py src/app/application/common/query_params/pagination.py renamed to src/app/application/common/query_params/offset_pagination.py

@@ -1,16 +1,12 @@
 import logging
 from dataclasses import dataclass
-from typing import TypedDict
 
-from app.application.common.exceptions.query import SortingError
-from app.application.common.ports.user_query_gateway import UserQueryGateway
-from app.application.common.query_models.user import UserQueryModel
-from app.application.common.query_params.pagination import Pagination
-from app.application.common.query_params.sorting import SortingOrder
-from app.application.common.query_params.user import (
-    UserListParams,
-    UserListSorting,
+from app.application.common.ports.user_query_gateway import (
+    ListUsersQM,
+    UserQueryGateway,
 )
+from app.application.common.query_params.offset_pagination import OffsetPaginationParams
+from app.application.common.query_params.sorting import SortingOrder, SortingParams
 from app.application.common.services.authorization.authorize import (
     authorize,
 )
@@ -26,16 +22,12 @@
 
 @dataclass(frozen=True, slots=True, kw_only=True)
 class ListUsersRequest:
-    limit: int
     offset: int
+    limit: int
     sorting_field: str
     sorting_order: SortingOrder
 
 
-class ListUsersResponse(TypedDict):
-    users: list[UserQueryModel]
-
-
 class ListUsersQueryService:
     """
     - Open to admins.
@@ -50,14 +42,14 @@ def __init__(
         self._current_user_service = current_user_service
         self._user_query_gateway = user_query_gateway
 
-    async def execute(self, request_data: ListUsersRequest) -> ListUsersResponse:
+    async def execute(self, request_data: ListUsersRequest) -> ListUsersQM:
         """
         :raises AuthenticationError:
         :raises DataMapperError:
         :raises AuthorizationError:
-        :raises ReaderError:
         :raises PaginationError:
         :raises SortingError:
+        :raises ReaderError:
         """
         log.info("List users: started.")
 
@@ -72,28 +64,18 @@ async def execute(self, request_data: ListUsersRequest) -> ListUsersResponse:
         )
 
         log.debug("Retrieving list of users.")
-        user_list_params = UserListParams(
-            pagination=Pagination(
-                limit=request_data.limit,
-                offset=request_data.offset,
-            ),
-            sorting=UserListSorting(
-                sorting_field=request_data.sorting_field,
-                sorting_order=request_data.sorting_order,
-            ),
+        pagination = OffsetPaginationParams(
+            limit=request_data.limit,
+            offset=request_data.offset,
         )
-
-        users: list[UserQueryModel] | None = await self._user_query_gateway.read_all(
-            user_list_params,
+        sorting = SortingParams(
+            field=request_data.sorting_field,
+            order=request_data.sorting_order,
+        )
+        response = await self._user_query_gateway.read_all(
+            pagination=pagination,
+            sorting=sorting,
         )
-        if users is None:
-            log.error(
-                "Retrieving list of users failed: invalid sorting column '%s'.",
-                request_data.sorting_field,
-            )
-            raise SortingError("Invalid sorting field.")
-
-        response = ListUsersResponse(users=users)
 
         log.info("List users: done.")
         return response

src/app/application/common/query_params/sorting.py

@@ -42,3 +42,6 @@ def __hash__(self) -> int:
         reduces the risk of hash collisions between different entity types.
         """
         return hash((type(self), self.id_))
+
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}(id_={self.id_!r})"

src/app/application/common/query_params/user.py

@@ -12,7 +12,7 @@
     DB_FLUSH_FAILED,
     DB_QUERY_FAILED,
 )
-from app.infrastructure.adapters.types import MainAsyncSession
+from app.infrastructure.adapters.types_ import MainAsyncSession
 from app.infrastructure.exceptions.gateway import DataMapperError
 
 log = logging.getLogger(__name__)