Add cursor pagination to admin lists
This commit is contained in:
@@ -1,11 +1,22 @@
|
||||
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
|
||||
@@ -18,11 +29,13 @@ class ListUsersCommand:
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user