damanukyan/adminka
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add order and purchase request data for analytics

Browse Source
This commit is contained in:
Sergey Kvachuk
2026-06-18 18:37:04 +03:00
parent 8ebe016591
commit 1247b79e0b
12 changed files with 611 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/application/abstractions/repositories/i_order_repository.py
View File

@@ -27,7 +27,12 @@ class IOrderRepository(ABC):
@abstractmethod
async def sum_by_date(self,*,year: int,month: int) -> Decimal:
async def sum_by_month(self,*,year: int,month: int, search: str | None = None) -> Decimal:
raise NotImplementedError
@abstractmethod
async def sum_by_day(self,*,year: int,month: int,day: int, search: str | None = None) -> Decimal:
raise NotImplementedError
@@ -37,10 +42,10 @@ class IOrderRepository(ABC):
@abstractmethod
async def count_by_date(self,*,year: int,month: int | None = None) -> int:
async def count_by_month(self,*,year: int,month: int | None = None, search: str | None = None) -> int:
raise NotImplementedError
@abstractmethod
async def count_by_user_id(self,*,user_id: str) -> int:
async def count_by_day(self,*,year: int,month: int,day: int, search: str | None = None) -> int:
raise NotImplementedError

23
src/application/abstractions/repositories/i_purchase_request_repository.py
View File

@@ -31,3 +31,26 @@ class IPurchaseRequestRepository(ABC):
@abstractmethod
async def count_all(self, *, status: str | None, organization_id: str | None) -> int:
raise NotImplementedError
@abstractmethod
async def count_for_month(
self,
*,
year: int,
month: int,
status: str | None,
organization_id: str | None
) -> int:
raise NotImplementedError
@abstractmethod
async def count_for_day(
self,
*,
year: int,
month: int,
day: int,
status: str | None,
organization_id: str | None
) -> int:
raise NotImplementedError

14
src/application/commands/__init__.py
View File

@@ -44,6 +44,14 @@ from src.application.commands.orders_commands import (
ListOrdersByDateCommand,
GetOrderCommand,
)
from src.application.commands.analytics_commands import (
GetPurchaseAnalyticsForYearCommand,
GetOrdersAnalyticsForYearCommand,
GetPurchaseAnalyticsForMonthCommand,
GetOrdersAnalyticsForMonthCommand,
GetPurchaseAnalyticsForWeekCommand,
GetOrdersAnalyticsForWeekCommand,
)
__all__ = [
'AdminLoginCommand',
@@ -78,4 +86,10 @@ __all__ = [
'ListUserOrdersCommand',
'ListOrdersByDateCommand',
'GetOrderCommand',
'GetPurchaseAnalyticsForYearCommand',
'GetOrdersAnalyticsForYearCommand',
'GetPurchaseAnalyticsForMonthCommand',
'GetOrdersAnalyticsForMonthCommand',
'GetPurchaseAnalyticsForWeekCommand',
'GetOrdersAnalyticsForWeekCommand',
]

239
src/application/commands/analytics_commands.py Normal file
View File

@@ -0,0 +1,239 @@
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from datetime import date, timedelta
from src.application.abstractions import IUnitOfWork
from src.application.contracts import ILogger
from src.application.domain.entities import OrderEntity, PurchaseRequestEntity
from src.application.domain.exceptions import NotFoundException
from src.infrastructure.database.decorators import transactional
YEAR=12
MONTH=30
WEEK=7
def _count_date(
year: int,
month: int | None = None,
day: int | None = None,
) -> tuple[int, int | None, int | None]:
if month is None:
return year - 1, None, None
if day is None:
return (year - 1, 12, None) if month == 1 else (year, month - 1, None)
prev_date = date(year, month, day) - timedelta(days=1)
return prev_date.year, prev_date.month, prev_date.day
class GetOrdersAnalyticsForYearCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
self._unit_of_work = unit_of_work
self._logger = logger
@transactional
async def __call__(
self,
*,
start_year: int,
start_month: int | None = None,
search: str | None = None,
) -> tuple[list[int], list[Decimal], int, Decimal, int]:
orders: list[int] = []
sums: list[Decimal] = []
year, month = start_year, start_month
for _ in range(YEAR):
order_count = await self._unit_of_work.order_repository.count_by_month(
year=year,
month=month,
search=search,
)
order_sum = await self._unit_of_work.order_repository.sum_by_month(
year=year,
month=month,
search=search,
)
orders.append(order_count)
sums.append(order_sum)
year, month, _ = _count_date(year, month)
return orders, sums, sum(orders), sum(sums), YEAR
class GetPurchaseAnalyticsForYearCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
self._unit_of_work = unit_of_work
self._logger = logger
@transactional
async def __call__(
self,
*,
start_year: int,
start_month: int | None = None,
status: str | None = None,
organization_id: str | None = None,
) -> tuple[list[int], int, int]:
requests: list[int] = []
year, month = start_year, start_month
for _ in range(YEAR):
request_count = await self._unit_of_work.purchase_request_repository.count_for_month(
year=year,
month=month,
status=status,
organization_id=organization_id,
)
requests.append(request_count)
year, month, _ = _count_date(year, month)
return requests, sum(requests), YEAR
class GetOrdersAnalyticsForMonthCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
self._unit_of_work = unit_of_work
self._logger = logger
@transactional
async def __call__(
self,
*,
start_year: int,
start_month: int,
start_day: int | None = None,
search: str | None = None,
) -> tuple[list[int], list[Decimal], int, Decimal, int]:
orders: list[int] = []
sums: list[Decimal] = []
year, month, day = start_year, start_month, start_day
for _ in range(MONTH):
order_count = await self._unit_of_work.order_repository.count_by_day(
year=year,
month=month,
day=day,
search=search,
)
order_sum = await self._unit_of_work.order_repository.sum_by_day(
year=year,
month=month,
day=day,
search=search,
)
orders.append(order_count)
sums.append(order_sum)
year, month, day = _count_date(year, month, day)
return orders, sums, sum(orders), sum(sums), MONTH
class GetPurchaseAnalyticsForMonthCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
self._unit_of_work = unit_of_work
self._logger = logger
@transactional
async def __call__(
self,
*,
start_year: int,
start_month: int,
start_day: int | None = None,
status: str | None = None,
organization_id: str | None = None,
) -> tuple[list[PurchaseRequestEntity], int, int]:
requests: list[int] = []
year, month, day = start_year, start_month, start_day
for _ in range(MONTH):
request_count = await self._unit_of_work.purchase_request_repository.count_for_day(
year=year,
month=month,
day=day,
status=status,
organization_id=organization_id,
)
requests.append(request_count)
year, month, day = _count_date(year, month, day)
return requests, sum(requests), MONTH
class GetPurchaseAnalyticsForWeekCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
self._unit_of_work = unit_of_work
self._logger = logger
@transactional
async def __call__(
self,
*,
start_year: int,
start_month: int,
start_day: int,
status: str | None = None,
organization_id: str | None = None,
) -> tuple[list[PurchaseRequestEntity], int, int]:
requests: list[int] = []
year, month, day = start_year, start_month, start_day
for _ in range(WEEK):
request_count = await self._unit_of_work.purchase_request_repository.count_for_day(
year=year,
month=month,
day=day,
status=status,
organization_id=organization_id,
)
requests.append(request_count)
year, month, day = _count_date(year, month, day)
return requests, sum(requests), WEEK
class GetOrdersAnalyticsForWeekCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
self._unit_of_work = unit_of_work
self._logger = logger
@transactional
async def __call__(
self,
*,
start_year: int,
start_month: int,
start_day: int,
search: str | None = None,
) -> tuple[list[OrderEntity], int, Decimal, int]:
orders: list[int] = []
sums: list[Decimal] = []
year, month, day = start_year, start_month, start_day
for _ in range(WEEK):
order_count = await self._unit_of_work.order_repository.count_by_day(
year=year,
month=month,
day=day,
search=search,
)
order_sum = await self._unit_of_work.order_repository.sum_by_day(
year=year,
month=month,
day=day,
search=search,
)
orders.append(order_count)
sums.append(order_sum)
year, month, day = _count_date(year, month, day)
return orders, sums, sum(orders), sum(sums), WEEK

15
src/application/commands/orders_commands.py
View File

@@ -21,7 +21,7 @@ class ListUserOrdersCommand:
limit=limit,
offset=offset,
)
total = await self._unit_of_work.order_repository.count_by_user_id(user_id=user_id)
total = await self._unit_of_work.order_repository.count_all(search=user_id)
return orders, total
@@ -52,7 +52,14 @@ class ListOrdersByDateCommand:
@transactional
async def __call__(self, *, limit: int, offset: int, year: int, month: int | None = None) -> tuple[list[OrderEntity], int, Decimal]:
async def __call__(
self,
*,
limit: int,
offset: int,
year: int,
month: int | None = None
) -> tuple[list[OrderEntity], int, Decimal]:
orders = await self._unit_of_work.order_repository.list_by_date(
year=year,
month=month,
@@ -60,8 +67,8 @@ class ListOrdersByDateCommand:
offset=offset
)
total = await self._unit_of_work.order_repository.count_by_date(year=year, month=month)
summary = await self._unit_of_work.order_repository.sum_by_date(year=year, month=month)
total = await self._unit_of_work.order_repository.count_by_month(year=year, month=month)
summary = await self._unit_of_work.order_repository.sum_by_month(year=year, month=month)
return orders, total, summary

4
src/application/domain/entities/__init__.py
View File

@@ -2,5 +2,7 @@ from src.application.domain.entities.user import UserEntity
from src.application.domain.entities.admin_user import AdminUserEntity
from src.application.domain.entities.admin_session import AdminSessionEntity
from src.application.domain.entities.order import OrderEntity
from src.application.domain.entities.organization import PurchaseRequestEntity
__all__ = ['UserEntity', 'AdminUserEntity', 'AdminSessionEntity', 'OrderEntity']
__all__ = ['UserEntity', 'AdminUserEntity', 'AdminSessionEntity', 'OrderEntity', 'PurchaseRequestEntity']

63
src/infrastructure/database/repositories/orders_repository.py
View File

@@ -52,6 +52,7 @@ class OrderRepository(IOrderRepository):
return or_(
OrderModel.id.ilike(pattern),
OrderModel.user_id.ilike(pattern),
OrderModel.status.ilike(pattern),
OrderModel.client_payment_id.ilike(pattern),
OrderModel.itpay_id.ilike(pattern),
OrderModel.itpay_qr_id.ilike(pattern),
@@ -72,7 +73,7 @@ class OrderRepository(IOrderRepository):
user_id: str,
limit: int,
offset: int
) -> list[OrderEntity]:
) -> list[OrderEntity]:
stmt=(
select(OrderModel)
.where(OrderModel.user_id==user_id)
@@ -90,7 +91,7 @@ class OrderRepository(IOrderRepository):
limit: int,
offset: int,
search: str | None = None
) -> list[OrderEntity]:
) -> list[OrderEntity]:
stmt=(
select(OrderModel)
.order_by(desc(OrderModel.created_at))
@@ -111,7 +112,7 @@ class OrderRepository(IOrderRepository):
month: int,
limit: int,
offset: int,
) -> list[OrderEntity]:
) -> list[OrderEntity]:
stmt=(
select(OrderModel)
.where(
@@ -126,7 +127,13 @@ class OrderRepository(IOrderRepository):
return [self._to_entity(model) for model in res.scalars().all()]
async def sum_by_date(self, *, year: int, month: int) -> Decimal:
async def sum_by_month(
self,
*,
year: int,
month: int,
search: str | None = None,
) -> Decimal:
stmt = (
select(func.coalesce(func.sum(OrderModel.service_fee), 0))
.where(
@@ -135,6 +142,33 @@ class OrderRepository(IOrderRepository):
OrderModel.status == OrderStatus.COMPLETED.value,
)
)
search_filter = self._search_filter(search)
if search_filter is not None:
stmt = stmt.where(search_filter)
res = await self._session.execute(stmt)
return res.scalar_one()
async def sum_by_day(
self,
*,
year: int,
month: int,
day: int,
search: str | None = None,
) -> Decimal:
stmt = (
select(func.coalesce(func.sum(OrderModel.service_fee), 0))
.where(
extract("year", OrderModel.created_at) == year,
extract("month", OrderModel.created_at) == month,
extract("day", OrderModel.created_at) == day,
OrderModel.status == OrderStatus.COMPLETED.value,
)
)
search_filter = self._search_filter(search)
if search_filter is not None:
stmt = stmt.where(search_filter)
res = await self._session.execute(stmt)
return res.scalar_one()
@@ -148,17 +182,28 @@ class OrderRepository(IOrderRepository):
return int(res.scalar_one())
async def count_by_date(self,*,year: int,month: int) -> int:
async def count_by_month(self,*,year: int,month: int, search: str | None = None) -> int:
stmt = select(func.count()).select_from(OrderModel)
stmt = stmt.where(
extract("year", OrderModel.created_at) == year,
extract("month", OrderModel.created_at) == month
)
search_filter = self._search_filter(search)
if search_filter is not None:
stmt = stmt.where(search_filter)
res = await self._session.execute(stmt)
return int(res.scalar_one())
async def count_by_user_id(self,*,user_id: str) -> int:
stmt = select(func.count()).select_from(OrderModel).where(OrderModel.user_id == user_id)
async def count_by_day(self,*,year: int,month: int,day: int, search: str | None = None) -> int:
stmt = select(func.count()).select_from(OrderModel)
stmt = stmt.where(
extract("year", OrderModel.created_at) == year,
extract("month", OrderModel.created_at) == month,
extract("day", OrderModel.created_at) == day
)
search_filter = self._search_filter(search)
if search_filter is not None:
stmt = stmt.where(search_filter)
res = await self._session.execute(stmt)
return int(res.scalar_one())
return int(res.scalar_one())

38
src/infrastructure/database/repositories/purchase_request_repository.py
View File

@@ -90,3 +90,41 @@ class PurchaseRequestRepository(IPurchaseRequestRepository):
)
await self._session.flush()
return await self.get_by_id(request_id)
async def count_for_month(
self,
*,
year: int,
month: int,
status: str | None,
organization_id: str | None
) -> int:
stmt = select(func.count()).select_from(PurchaseRequestModel)
stmt = self._apply_filters(stmt, status=status, organization_id=organization_id)
stmt = stmt.where(
func.extract("year", PurchaseRequestModel.created_at) == year,
func.extract("month", PurchaseRequestModel.created_at) == month,
)
res = await self._session.execute(stmt)
return int(res.scalar_one())
async def count_for_day(
self,
*,
year: int,
month: int,
day: int,
status: str | None,
organization_id: str | None
) -> int:
stmt = select(func.count()).select_from(PurchaseRequestModel)
stmt = self._apply_filters(stmt, status=status, organization_id=organization_id)
stmt = stmt.where(
func.extract("year", PurchaseRequestModel.created_at) == year,
func.extract("month", PurchaseRequestModel.created_at) == month,
func.extract("day", PurchaseRequestModel.created_at) == day,
)
res = await self._session.execute(stmt)
return int(res.scalar_one())

44
src/presentation/dependencies/commands.py
View File

@@ -36,6 +36,12 @@ from src.application.commands import (
ListUserOrdersCommand,
ListOrdersByDateCommand,
GetOrderCommand,
GetPurchaseAnalyticsForYearCommand,
GetOrdersAnalyticsForYearCommand,
GetPurchaseAnalyticsForMonthCommand,
GetOrdersAnalyticsForMonthCommand,
GetPurchaseAnalyticsForWeekCommand,
GetOrdersAnalyticsForWeekCommand,
)
from src.application.contracts import IHashService, IJwtService, ILogger
from src.infrastructure.config import settings
@@ -277,4 +283,40 @@ def get_get_order_command(
uow: IUnitOfWork = Depends(get_unit_of_work),
logger: ILogger = Depends(get_logger),
) -> GetOrderCommand:
return GetOrderCommand(uow, logger)
return GetOrderCommand(uow, logger)
def get_get_purchase_analytics_for_year_command(
uow: IUnitOfWork = Depends(get_unit_of_work),
logger: ILogger = Depends(get_logger),
) -> GetPurchaseAnalyticsForYearCommand:
return GetPurchaseAnalyticsForYearCommand(uow, logger)
def get_get_orders_analytics_for_year_command(
uow: IUnitOfWork = Depends(get_unit_of_work),
logger: ILogger = Depends(get_logger),
) -> GetOrdersAnalyticsForYearCommand:
return GetOrdersAnalyticsForYearCommand(uow, logger)
def get_get_purchase_analytics_for_month_command(
uow: IUnitOfWork = Depends(get_unit_of_work),
logger: ILogger = Depends(get_logger),
) -> GetPurchaseAnalyticsForMonthCommand:
return GetPurchaseAnalyticsForMonthCommand(uow, logger)
def get_get_orders_analytics_for_month_command(
uow: IUnitOfWork = Depends(get_unit_of_work),
logger: ILogger = Depends(get_logger),
) -> GetOrdersAnalyticsForMonthCommand:
return GetOrdersAnalyticsForMonthCommand(uow, logger)
def get_get_purchase_analytics_for_week_command(
uow: IUnitOfWork = Depends(get_unit_of_work),
logger: ILogger = Depends(get_logger),
) -> GetPurchaseAnalyticsForWeekCommand:
return GetPurchaseAnalyticsForWeekCommand(uow, logger)
def get_get_orders_analytics_for_week_command(
uow: IUnitOfWork = Depends(get_unit_of_work),
logger: ILogger = Depends(get_logger),
) -> GetOrdersAnalyticsForWeekCommand:
return GetOrdersAnalyticsForWeekCommand(uow, logger)

4
src/presentation/routing/__init__.py
View File

@@ -8,6 +8,7 @@ from src.presentation.routing.purchase_requests import purchase_requests_router
from src.presentation.routing.users import users_router
from src.presentation.routing.search import search_router
from src.presentation.routing.orders import orders_router
from src.presentation.routing.analytics import analytics_router
v1_router = APIRouter(prefix='/v1')
v1_router.include_router(auth_router)
@@ -17,4 +18,5 @@ v1_router.include_router(documents_router)
v1_router.include_router(purchase_requests_router)
v1_router.include_router(users_router)
v1_router.include_router(search_router)
v1_router.include_router(orders_router)
v1_router.include_router(orders_router)
v1_router.include_router(analytics_router)

160
src/presentation/routing/analytics.py Normal file
View File

@@ -0,0 +1,160 @@
from fastapi import APIRouter, Depends, Request, Query, status
from src.application.domain.dto import AdminAuthContext
from src.presentation.decorators.admin_auth import require_admin_access, require_admin_role
from src.presentation.dependencies.commands import (
get_get_purchase_analytics_for_year_command,
get_get_orders_analytics_for_year_command,
get_get_purchase_analytics_for_month_command,
get_get_orders_analytics_for_month_command,
get_get_purchase_analytics_for_week_command,
get_get_orders_analytics_for_week_command,
)
from src.presentation.schemas.analytics import (
OrderAnalyticsResponse,
PurchaseRequestAnalyticsResponse,
)
from src.application.commands.analytics_commands import (
GetPurchaseAnalyticsForYearCommand,
GetOrdersAnalyticsForYearCommand,
GetPurchaseAnalyticsForMonthCommand,
GetOrdersAnalyticsForMonthCommand,
GetPurchaseAnalyticsForWeekCommand,
GetOrdersAnalyticsForWeekCommand,
)
analytics_router = APIRouter(prefix='/analytics', tags=['analytics'])
@analytics_router.get('/purchases/yearly', response_model=PurchaseRequestAnalyticsResponse)
async def get_purchase_analytics_for_year(
start_year: int = Query(..., ge=2000, le=2100),
start_month: int = Query(..., ge=1, le=12),
auth: AdminAuthContext = Depends(require_admin_access),
status: str | None = Query(default=None, min_length=1, max_length=255),
organization_id: str | None = Query(default=None, min_length=1, max_length=255),
command: GetPurchaseAnalyticsForYearCommand = Depends(get_get_purchase_analytics_for_year_command),
):
items, total, period = await command(
start_year=start_year,
start_month=start_month,
status=status,
organization_id=organization_id,
)
return PurchaseRequestAnalyticsResponse(
requests_count=items,
total_requests=total,
period=period,
)
@analytics_router.get('/orders/yearly', response_model=OrderAnalyticsResponse)
async def get_orders_analytics_for_year(
start_year: int = Query(..., ge=2000, le=2100),
start_month: int = Query(..., ge=1, le=12),
auth: AdminAuthContext = Depends(require_admin_access),
q: str | None = Query(default=None, min_length=1, max_length=255),
command: GetOrdersAnalyticsForYearCommand = Depends(get_get_orders_analytics_for_year_command),
):
orders_count, orders_sum, total_orders, total_sum, period = await command(
start_year=start_year,
start_month=start_month,
search=q,
)
return OrderAnalyticsResponse(
orders_count=orders_count,
summed_service_fees=orders_sum,
total_orders=total_orders,
total_summed_service_fees=total_sum,
period=period,
)
@analytics_router.get('/purchases/monthly', response_model=PurchaseRequestAnalyticsResponse)
async def get_purchase_analytics_for_month(
start_year: int = Query(..., ge=2000, le=2100),
start_month: int = Query(..., ge=1, le=12),
start_day: int = Query(..., ge=1, le=31),
auth: AdminAuthContext = Depends(require_admin_access),
status: str | None = Query(default=None, min_length=1, max_length=255),
organization_id: str | None = Query(default=None, min_length=1, max_length=255),
command: GetPurchaseAnalyticsForMonthCommand = Depends(get_get_purchase_analytics_for_month_command),
):
items, total, period = await command(
start_year=start_year,
start_month=start_month,
start_day=start_day,
status=status,
organization_id=organization_id,
)
return PurchaseRequestAnalyticsResponse(
requests_count=items,
total_requests=total,
period=period,
)
@analytics_router.get('/orders/monthly', response_model=OrderAnalyticsResponse)
async def get_orders_analytics_for_month(
start_year: int = Query(..., ge=2000, le=2100),
start_month: int = Query(..., ge=1, le=12),
start_day: int = Query(..., ge=1, le=31),
auth: AdminAuthContext = Depends(require_admin_access),
q: str | None = Query(default=None, min_length=1, max_length=255),
command: GetOrdersAnalyticsForMonthCommand = Depends(get_get_orders_analytics_for_month_command),
):
orders_count, orders_sum, total_orders, total_sum, period = await command(
start_year=start_year,
start_month=start_month,
start_day=start_day,
search=q,
)
return OrderAnalyticsResponse(
orders_count=orders_count,
summed_service_fees=orders_sum,
total_orders=total_orders,
total_summed_service_fees=total_sum,
period=period,
)
@analytics_router.get('/purchases/weekly', response_model=PurchaseRequestAnalyticsResponse)
async def get_purchase_analytics_for_week(
start_year: int = Query(..., ge=2000, le=2100),
start_month: int = Query(..., ge=1, le=12),
start_day: int = Query(..., ge=1, le=31),
auth: AdminAuthContext = Depends(require_admin_access),
status: str | None = Query(default=None, min_length=1, max_length=255),
organization_id: str | None = Query(default=None, min_length=1, max_length=255),
command: GetPurchaseAnalyticsForWeekCommand = Depends(get_get_purchase_analytics_for_week_command),
):
items, total, period = await command(
start_year=start_year,
start_month=start_month,
start_day=start_day,
status=status,
organization_id=organization_id,
)
return PurchaseRequestAnalyticsResponse(
requests_count=items,
total_requests=total,
period=period,
)
@analytics_router.get('/orders/weekly', response_model=OrderAnalyticsResponse)
async def get_orders_analytics_for_week(
start_year: int = Query(..., ge=2000, le=2100),
start_month: int = Query(..., ge=1, le=12),
start_day: int = Query(..., ge=1, le=31),
auth: AdminAuthContext = Depends(require_admin_access),
q: str | None = Query(default=None, min_length=1, max_length=255),
command: GetOrdersAnalyticsForWeekCommand = Depends(get_get_orders_analytics_for_week_command),
):
orders_count, orders_sum, total_orders, total_sum, period = await command(
start_year=start_year,
start_month=start_month,
start_day=start_day,
search=q,
)
return OrderAnalyticsResponse(
orders_count=orders_count,
summed_service_fees=orders_sum,
total_orders=total_orders,
total_summed_service_fees=total_sum,
period=period,
)

15
src/presentation/schemas/analytics.py Normal file
View File

@@ -0,0 +1,15 @@
from decimal import Decimal
from pydantic import BaseModel
class OrderAnalyticsResponse(BaseModel):
orders_count: list[int]
summed_service_fees: list[Decimal]
total_orders: int
total_summed_service_fees: Decimal
period: int | None = None
class PurchaseRequestAnalyticsResponse(BaseModel):
requests_count: list[int]
total_requests: int
period: int | None = None