52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from src.application.abstractions import IUnitOfWork
|
|
from src.application.contracts import ILogger
|
|
from src.application.domain.entities.user import UserEntity
|
|
from src.infrastructure.database.decorators import transactional
|
|
|
|
|
|
def _parse_cursor(cursor: str | None) -> datetime | None:
|
|
if not cursor:
|
|
return None
|
|
parsed = datetime.fromisoformat(cursor.replace('Z','+00:00'))
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=timezone.utc)
|
|
return parsed
|
|
|
|
|
|
class ListUsersCommand:
|
|
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
|
|
self._unit_of_work = unit_of_work
|
|
self._logger = logger
|
|
|
|
@transactional
|
|
async def __call__(
|
|
self,
|
|
*,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
search: str | None = None,
|
|
cursor: str | None = None,
|
|
) -> tuple[list[UserEntity], int]:
|
|
items = await self._unit_of_work.user_repository.list_all(
|
|
limit=limit,
|
|
offset=offset,
|
|
search=search,
|
|
cursor_created_at=_parse_cursor(cursor),
|
|
)
|
|
total = await self._unit_of_work.user_repository.count_all(search=search)
|
|
return items, total
|
|
|
|
|
|
class GetUserCommand:
|
|
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
|
|
self._unit_of_work = unit_of_work
|
|
self._logger = logger
|
|
|
|
@transactional
|
|
async def __call__(self, user_id: str) -> UserEntity:
|
|
return await self._unit_of_work.user_repository.get_by_id(user_id)
|