120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal
|
|
from ulid import ULID
|
|
|
|
from src.application.domain.exceptions.application_exceptions import ApplicationException
|
|
from src.infrastructure.config import settings
|
|
from src.infrastructure.context_vars import trace_id_var
|
|
from src.application.abstractions import IUnitOfWork
|
|
from src.application.commands.legal_entity_guard import require_legal_entity
|
|
from src.application.contracts import ILogger, IQueueMessanger
|
|
from src.application.domain.entities.purchase_request import PurchaseRequestEntity
|
|
from src.application.domain.exceptions import NotFoundException
|
|
from src.infrastructure.database.decorators import transactional
|
|
|
|
|
|
class CreatePurchaseRequestCommand:
|
|
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger, messanger: IQueueMessanger):
|
|
self._unit_of_work = unit_of_work
|
|
self._logger = logger
|
|
self._messanger = messanger
|
|
|
|
@transactional
|
|
async def __call__(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
rub_amount: Decimal,
|
|
comment: str | None = None,
|
|
target_wallet_chain: str | None = None,
|
|
target_wallet_address: str | None = None,
|
|
) -> PurchaseRequestEntity:
|
|
legal_entity = await require_legal_entity(user_id, self._unit_of_work)
|
|
item = await self._unit_of_work.purchase_request_repository.create(
|
|
organization_id=legal_entity.id,
|
|
rub_amount=rub_amount,
|
|
comment=comment,
|
|
target_wallet_chain=target_wallet_chain,
|
|
target_wallet_address=target_wallet_address,
|
|
)
|
|
trace_id = trace_id_var.get()
|
|
if not trace_id or trace_id == 'N/A':
|
|
trace_id = None
|
|
|
|
message_id = str(ULID())
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
metadata = {
|
|
'trace_id': trace_id,
|
|
'timestamp': now,
|
|
'message_id': message_id,
|
|
}
|
|
payload = {
|
|
'id_client': legal_entity.id,
|
|
'rub_amount': rub_amount,
|
|
}
|
|
message = {
|
|
'event': 'purchase_created',
|
|
'payload': payload,
|
|
'metadata': metadata,
|
|
}
|
|
self._logger.info(f'Create purchase request for organization_id={legal_entity.id } with rub_amount={rub_amount}')
|
|
try:
|
|
await self._messanger.publish_to_queue(
|
|
queue=settings.RABBIT_TELEGRAM_NOTIFY_QUEUE,
|
|
message=message,
|
|
persist=True,
|
|
correlation_id=trace_id,
|
|
message_id=message_id,
|
|
headers={'trace_id': trace_id} if trace_id else None,
|
|
)
|
|
except Exception as exception:
|
|
self._logger.error(f'Failed to publish purchase request notification for organization_id={legal_entity.id}: {str(exception)}')
|
|
raise ApplicationException(503, 'Temporary error. Please try again.')
|
|
|
|
self._logger.info(f'Purchase request created id= organization_id={legal_entity.id }')
|
|
return item
|
|
|
|
|
|
class ListPurchaseRequestsCommand:
|
|
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,
|
|
*,
|
|
status: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> tuple[list[PurchaseRequestEntity], int]:
|
|
legal_entity = await require_legal_entity(user_id, self._unit_of_work)
|
|
items = await self._unit_of_work.purchase_request_repository.list_by_organization(
|
|
organization_id=legal_entity.id,
|
|
status=status,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
total = await self._unit_of_work.purchase_request_repository.count_by_organization(
|
|
organization_id=legal_entity.id,
|
|
status=status,
|
|
)
|
|
return items, total
|
|
|
|
|
|
class GetPurchaseRequestCommand:
|
|
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, request_id: str) -> PurchaseRequestEntity:
|
|
legal_entity = await require_legal_entity(user_id, self._unit_of_work)
|
|
item = await self._unit_of_work.purchase_request_repository.get_by_id(request_id)
|
|
if item.organization_id != legal_entity.id:
|
|
raise NotFoundException(message='Purchase request not found')
|
|
return item
|