damanukyan/pay-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add withdraw

Browse Source
This commit is contained in:
Daniil Manukian
2026-06-21 14:16:20 +03:00
parent 8f60637e8d
commit 56841b83ce
32 changed files with 1291 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
docker-compose.yml
View File

@@ -14,4 +14,44 @@ services:
APP_WORKERS: "2"
env_file:
- .env
depends_on:
pay_keydb:
condition: service_healthy
restart: no
pay_keydb:
image: eqalpha/keydb
container_name: pay_keydb
restart: no
expose:
- "6379"
volumes:
- pay_keydb_data:/data
command:
- keydb-server
- --requirepass
- ${REDIS_PASSWORD}
- --dir
- /data
- --appendonly
- "yes"
- --appendfsync
- everysec
- --save
- "900"
- "1"
- --save
- "300"
- "10"
- --save
- "60"
- "10000"
healthcheck:
test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"]
interval: 5s
timeout: 2s
retries: 20
volumes:
pay_keydb_data:

5
src/application/abstractions/i_unit_of_work.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Protocol, runtime_checkable
from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,IUserRepository
from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,ISbpWithdrawalRepository,IUserRepository
@runtime_checkable
@@ -17,6 +17,9 @@ class IUnitOfWork(Protocol):
@property
def payment_repository(self) -> IPaymentRepository: ...
@property
def sbp_withdrawal_repository(self) -> ISbpWithdrawalRepository: ...
@property
def user_repository(self) -> IUserRepository: ...

1
src/application/abstractions/repositories/__init__.py
View File

@@ -1,3 +1,4 @@
from src.application.abstractions.repositories.i_order_repository import IOrderRepository
from src.application.abstractions.repositories.i_payment_repository import IPaymentRepository
from src.application.abstractions.repositories.i_sbp_withdrawal_repository import ISbpWithdrawalRepository
from src.application.abstractions.repositories.i_user_repository import IUserRepository

47
src/application/abstractions/repositories/i_sbp_withdrawal_repository.py Normal file
View File

@@ -0,0 +1,47 @@
from abc import ABC,abstractmethod
from src.application.domain.entities import SbpWithdrawalEntity
from src.application.domain.enums import SbpWithdrawalStatus
class ISbpWithdrawalRepository(ABC):
@abstractmethod
async def create(self,withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalEntity:
raise NotImplementedError
@abstractmethod
async def get_by_id(self,withdraw_id: str) -> SbpWithdrawalEntity | None:
raise NotImplementedError
@abstractmethod
async def get_by_id_for_user(self,*,withdraw_id: str,user_id: str) -> SbpWithdrawalEntity | None:
raise NotImplementedError
@abstractmethod
async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[SbpWithdrawalEntity]:
raise NotImplementedError
@abstractmethod
async def update_status(
self,
*,
withdraw_id: str,
status: SbpWithdrawalStatus,
provider_error: str | None = None,
) -> None:
raise NotImplementedError
@abstractmethod
async def update_payout_created(
self,
*,
withdraw_id: str,
mozen_order_id: str | None,
status: SbpWithdrawalStatus,
provider_error: str | None,
) -> None:
raise NotImplementedError

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

@@ -1,4 +1,5 @@
from src.application.commands.create_order_command import CreateOrderCommand
from src.application.commands.create_payment_command import CreatePaymentCommand
from src.application.commands.create_crypto_transfer_completed_command import CreateCryptoTransferCompletedCommand
from src.application.commands.payment_read_commands import GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListOrdersCommand,ListPaymentsCommand,OrderPaymentResult
from src.application.commands.payment_read_commands import ClientOperationResult,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListClientOperationsCommand,ListOrdersCommand,ListPaymentsCommand,OrderPaymentResult
from src.application.commands.sbp_withdrawal_commands import CreateSbpWithdrawalCommand,GetSbpBanksCommand,GetSbpWithdrawalCommand,HandleSbpWithdrawalWalletEventCommand

45
src/application/commands/payment_read_commands.py
View File

@@ -1,9 +1,11 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime,timezone
from decimal import Decimal
from typing import Literal
from src.application.abstractions import IUnitOfWork
from src.application.contracts import ILogger
from src.application.domain.entities import OrderEntity,PaymentEntity
from src.application.domain.entities import OrderEntity,PaymentEntity,SbpWithdrawalEntity
from src.application.domain.exceptions import NotFoundException
from src.application.services import PaymentQuote,PaymentQuoteService
from src.infrastructure.database.decorators import transactional
@@ -15,6 +17,14 @@ class OrderPaymentResult:
payment: PaymentEntity | None
@dataclass(slots=True)
class ClientOperationResult:
type: Literal['payment','withdrawal']
created_at: datetime
payment: PaymentEntity | None = None
withdrawal: SbpWithdrawalEntity | None = None
class GetPaymentConfigCommand:
def __init__(self, *, payment_quote_service: PaymentQuoteService, logger: ILogger):
self._payment_quote_service = payment_quote_service
@@ -91,6 +101,39 @@ class ListPaymentsCommand:
return payments
class ListClientOperationsCommand:
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, limit: int, offset: int) -> list[ClientOperationResult]:
fetch_limit=limit+offset
payments=await self._unit_of_work.payment_repository.list_by_user_id(
user_id=user_id,
limit=fetch_limit,
offset=0,
)
withdrawals=await self._unit_of_work.sbp_withdrawal_repository.list_by_user_id(
user_id=user_id,
limit=fetch_limit,
offset=0,
)
min_dt=datetime.min.replace(tzinfo=timezone.utc)
items=[
ClientOperationResult(type='payment',created_at=payment.created_at or min_dt,payment=payment)
for payment in payments
]
items.extend(
ClientOperationResult(type='withdrawal',created_at=withdrawal.created_at or min_dt,withdrawal=withdrawal)
for withdrawal in withdrawals
)
items.sort(key=lambda item:item.created_at,reverse=True)
self._logger.info({'event':'client_operations_list_requested','user_id':user_id,'limit':limit,'offset':offset})
return items[offset:offset+limit]
class GetPaymentCommand:
def __init__(self, *, unit_of_work: IUnitOfWork, logger: ILogger):
self._unit_of_work = unit_of_work

254
src/application/commands/sbp_withdrawal_commands.py Normal file
View File

@@ -0,0 +1,254 @@
from __future__ import annotations
from datetime import datetime,timezone
from decimal import Decimal
from typing import Any
from ulid import ULID
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IQueueMessanger,ILogger,IMozenSbpService,SbpBank
from src.application.domain.entities import SbpWithdrawalEntity
from src.application.domain.enums import SbpWithdrawalStatus
from src.application.domain.exceptions import BadRequestException,NotFoundException,PaymentProviderException,ServiceUnavailableException
from src.application.services import SbpWithdrawalQuoteService
from src.infrastructure.config import settings
from src.infrastructure.database.decorators import transactional
class GetSbpBanksCommand:
def __init__(self,*,mozen_sbp_service: IMozenSbpService,logger: ILogger):
self._mozen_sbp_service=mozen_sbp_service
self._logger=logger
async def __call__(self) -> list[SbpBank]:
return await self._mozen_sbp_service.get_banks(self._logger.get_trace_id())
class CreateSbpWithdrawalCommand:
def __init__(
self,
*,
unit_of_work: IUnitOfWork,
logger: ILogger,
queue_messanger: IQueueMessanger,
mozen_sbp_service: IMozenSbpService,
quote_service: SbpWithdrawalQuoteService,
):
self._unit_of_work=unit_of_work
self._logger=logger
self._queue_messanger=queue_messanger
self._mozen_sbp_service=mozen_sbp_service
self._quote_service=quote_service
async def _bank_name(self,bank_id: str) -> str:
banks=await self._mozen_sbp_service.get_banks(self._logger.get_trace_id())
for bank in banks:
if bank.id==bank_id:
return bank.name
raise BadRequestException(message='SBP bank not found')
@transactional
async def __call__(self,*,user_id: str,bank_id: str,phone: str,usdt_amount: Decimal) -> SbpWithdrawalEntity:
user=await self._unit_of_work.user_repository.get(user_id)
if user is None:
raise NotFoundException(message='User not found')
wallet_address=str(settings.SBP_WITHDRAWAL_WALLET_ADDRESS or '').strip()
if not wallet_address:
raise ServiceUnavailableException(message='SBP withdrawal wallet address unavailable')
trace_id=self._logger.get_trace_id()
withdraw_id=str(ULID())
bank_name=await self._bank_name(bank_id)
quote=await self._quote_service.get_quote(usdt_amount)
withdrawal=SbpWithdrawalEntity(
id=withdraw_id,
user_id=user_id,
bank_id=bank_id,
bank_name=bank_name,
phone=phone,
wallet_address=wallet_address,
usdt_amount=quote.usdt_amount,
rub_amount=quote.rub_amount,
usdt_exchange_rate=quote.usdt_exchange_rate,
service_fee_rate=quote.service_fee_rate,
service_fee_usdt=quote.service_fee_usdt,
trace_id=trace_id,
status=SbpWithdrawalStatus.WAITING_USDT,
created_at=datetime.now(timezone.utc),
)
saved=await self._unit_of_work.sbp_withdrawal_repository.create(withdrawal)
message_id=str(ULID())
message: dict[str,Any]={
'user_id':user_id,
'usdt_amount':str(quote.usdt_amount),
'trace_id':trace_id,
'wallet_address':wallet_address,
'withdraw_id':withdraw_id,
'message_id':message_id,
}
await self._queue_messanger.publish_to_queue(
queue=settings.RABBIT_SBP_WITHDRAWAL_USDT_REQUESTED_QUEUE,
message=message,
message_id=message_id,
correlation_id=trace_id,
headers={'trace_id':trace_id},
)
self._logger.info({
'event':'sbp_withdrawal_usdt_request_published',
'withdraw_id':withdraw_id,
'user_id':user_id,
'message_id':message_id,
})
return saved
class GetSbpWithdrawalCommand:
def __init__(self,*,unit_of_work: IUnitOfWork,logger: ILogger):
self._unit_of_work=unit_of_work
self._logger=logger
@transactional
async def __call__(self,*,withdraw_id: str,user_id: str) -> SbpWithdrawalEntity:
withdrawal=await self._unit_of_work.sbp_withdrawal_repository.get_by_id_for_user(
withdraw_id=withdraw_id,
user_id=user_id,
)
if withdrawal is None:
raise NotFoundException(message='SBP withdrawal not found')
return withdrawal
class HandleSbpWithdrawalWalletEventCommand:
def __init__(
self,
*,
unit_of_work: IUnitOfWork,
logger: ILogger,
mozen_sbp_service: IMozenSbpService,
):
self._unit_of_work=unit_of_work
self._logger=logger
self._mozen_sbp_service=mozen_sbp_service
@staticmethod
def _event_error(
*,
event_type: str,
received_usdt_amount: Decimal | None,
tx_hash: str | None,
error: str | None,
) -> str | None:
parts=(
f'event_type={event_type}',
f'received_usdt_amount={received_usdt_amount}' if received_usdt_amount is not None else '',
f'tx_hash={tx_hash}' if tx_hash else '',
f'error={error}' if error else '',
)
value=';'.join(part for part in parts if part)
return value or None
@transactional
async def __call__(
self,
*,
withdraw_id: str,
user_id: str,
event_type: str,
received_usdt_amount: Decimal | None = None,
tx_hash: str | None = None,
error: str | None = None,
) -> None:
withdrawal=await self._unit_of_work.sbp_withdrawal_repository.get_by_id(withdraw_id)
if withdrawal is None:
raise NotFoundException(message='SBP withdrawal not found')
if withdrawal.user_id!=user_id:
raise BadRequestException(message='SBP withdrawal user mismatch')
normalized_event_type=event_type.strip().lower()
if withdrawal.status in (
SbpWithdrawalStatus.PAYOUT_PROCESSING,
SbpWithdrawalStatus.PAYOUT_COMPLETED,
SbpWithdrawalStatus.PAYOUT_FAILED,
SbpWithdrawalStatus.CANCELLED,
SbpWithdrawalStatus.EXPIRED,
):
self._logger.info({
'event':'sbp_withdrawal_wallet_event_ignored',
'withdraw_id':withdraw_id,
'user_id':user_id,
'wallet_event_type':normalized_event_type,
'status':withdrawal.status.value if withdrawal.status is not None else None,
})
return
provider_error=self._event_error(
event_type=normalized_event_type,
received_usdt_amount=received_usdt_amount,
tx_hash=tx_hash,
error=error,
)
if normalized_event_type in ('usdt_underpaid','underpaid'):
await self._unit_of_work.sbp_withdrawal_repository.update_status(
withdraw_id=withdraw_id,
status=SbpWithdrawalStatus.USDT_UNDERPAID,
provider_error=provider_error,
)
return
if normalized_event_type in ('expired','usdt_expired'):
await self._unit_of_work.sbp_withdrawal_repository.update_status(
withdraw_id=withdraw_id,
status=SbpWithdrawalStatus.EXPIRED,
provider_error=provider_error,
)
return
if normalized_event_type in ('failed','wallet_error','watch_failed','error'):
await self._unit_of_work.sbp_withdrawal_repository.update_status(
withdraw_id=withdraw_id,
status=SbpWithdrawalStatus.WALLET_ERROR,
provider_error=provider_error,
)
return
if normalized_event_type not in ('usdt_received','received','confirmed'):
raise BadRequestException(message='Unknown SBP withdrawal wallet event type')
await self._unit_of_work.sbp_withdrawal_repository.update_status(
withdraw_id=withdraw_id,
status=SbpWithdrawalStatus.USDT_RECEIVED,
provider_error=provider_error,
)
try:
result=await self._mozen_sbp_service.make_withdrawal(
user_id=user_id,
withdraw_id=withdraw_id,
phone=str(withdrawal.phone or ''),
bank_id=str(withdrawal.bank_id or ''),
# amount=Decimal(str(withdrawal.rub_amount or '0')),
amount=Decimal('1'),
trace_id=str(withdrawal.trace_id or self._logger.get_trace_id()),
)
except PaymentProviderException as exception:
await self._unit_of_work.sbp_withdrawal_repository.update_status(
withdraw_id=withdraw_id,
status=SbpWithdrawalStatus.PAYOUT_FAILED,
provider_error=exception.message,
)
return
status=SbpWithdrawalStatus.PAYOUT_PROCESSING
provider_error=result.error
if result.error:
status=SbpWithdrawalStatus.PAYOUT_FAILED
elif result.status.strip().lower() in ('completed','success','succeeded','paid'):
status=SbpWithdrawalStatus.PAYOUT_COMPLETED
await self._unit_of_work.sbp_withdrawal_repository.update_payout_created(
withdraw_id=withdraw_id,
mozen_order_id=result.mozen_order_id,
status=status,
provider_error=provider_error,
)
self._logger.info({
'event':'sbp_withdrawal_payout_created',
'withdraw_id':withdraw_id,
'user_id':user_id,
'mozen_order_id':result.mozen_order_id,
'status':status.value,
})

1
src/application/contracts/__init__.py
View File

@@ -6,3 +6,4 @@ from src.application.contracts.i_hash_service import IHashService
from src.application.contracts.i_queue_messanger import IQueueMessanger
from src.application.contracts.i_itpay_service import IItPayService
from src.application.contracts.i_receipt import IReceipt
from src.application.contracts.i_mozen_sbp_service import IMozenSbpService,SbpBank,SbpPayoutResult

36
src/application/contracts/i_mozen_sbp_service.py Normal file
View File

@@ -0,0 +1,36 @@
from abc import ABC,abstractmethod
from dataclasses import dataclass
from decimal import Decimal
@dataclass(slots=True)
class SbpBank:
id: str
name: str
@dataclass(slots=True)
class SbpPayoutResult:
status: str
mozen_order_id: str | None = None
error: str | None = None
class IMozenSbpService(ABC):
@abstractmethod
async def get_banks(self,trace_id: str) -> list[SbpBank]:
raise NotImplementedError
@abstractmethod
async def make_withdrawal(
self,
*,
user_id: str,
withdraw_id: str,
phone: str,
bank_id: str,
amount: Decimal,
trace_id: str,
) -> SbpPayoutResult:
raise NotImplementedError

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

@@ -1,5 +1,6 @@
from src.application.domain.entities.order import OrderEntity
from src.application.domain.entities.payment import PaymentEntity
from src.application.domain.entities.sbp_withdrawal import SbpWithdrawalEntity
__all__ = ['PaymentEntity', 'OrderEntity']
__all__ = ['PaymentEntity','OrderEntity','SbpWithdrawalEntity']

30
src/application/domain/entities/sbp_withdrawal.py Normal file
View File

@@ -0,0 +1,30 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from src.application.domain.enums import SbpWithdrawalStatus
@dataclass(slots=True)
class SbpWithdrawalEntity:
id: str | None = None
created_at: datetime | None = None
updated_at: datetime | None = None
user_id: str | None = None
bank_id: str | None = None
bank_name: str | None = None
phone: str | None = None
wallet_address: str | None = None
usdt_amount: Decimal | None = None
rub_amount: Decimal | None = None
usdt_exchange_rate: Decimal | None = None
service_fee_rate: Decimal | None = None
service_fee_usdt: Decimal | None = None
mozen_order_id: str | None = None
trace_id: str | None = None
status: SbpWithdrawalStatus | None = None
provider_error: str | None = None
usdt_received_at: datetime | None = None
payout_created_at: datetime | None = None
payout_completed_at: datetime | None = None

1
src/application/domain/enums/__init__.py
View File

@@ -3,3 +3,4 @@ from src.application.domain.enums.log_format import LogFormat
from src.application.domain.enums.itpay_payment_status import ItPayPaymentStatus
from src.application.domain.enums.order_status import OrderStatus
from src.application.domain.enums.payment_status import PaymentStatus
from src.application.domain.enums.sbp_withdrawal_status import SbpWithdrawalStatus

14
src/application/domain/enums/sbp_withdrawal_status.py Normal file
View File

@@ -0,0 +1,14 @@
from enum import Enum
class SbpWithdrawalStatus(str,Enum):
CREATED='created'
WAITING_USDT='waiting_usdt'
USDT_UNDERPAID='usdt_underpaid'
USDT_RECEIVED='usdt_received'
WALLET_ERROR='wallet_error'
PAYOUT_PROCESSING='payout_processing'
PAYOUT_COMPLETED='payout_completed'
PAYOUT_FAILED='payout_failed'
CANCELLED='cancelled'
EXPIRED='expired'

1
src/application/services/__init__.py
View File

@@ -1,4 +1,5 @@
from src.application.services.payment_quote_service import PaymentQuote,PaymentQuoteService
from src.application.services.sbp_withdrawal_quote_service import SbpWithdrawalQuote,SbpWithdrawalQuoteService
__all__ = ['PaymentQuote', 'PaymentQuoteService']

64
src/application/services/sbp_withdrawal_quote_service.py Normal file
View File

@@ -0,0 +1,64 @@
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal,ROUND_DOWN,ROUND_UP
from src.application.contracts import ICache,ILogger
from src.application.domain.exceptions import OrderTotalOutOfRangeException,ServiceUnavailableException
MIN_SBP_WITHDRAWAL_USDT: Decimal=Decimal('100')
MAX_SBP_WITHDRAWAL_USDT: Decimal=Decimal('5000')
_FEE_TIERS: tuple[tuple[Decimal,Decimal,Decimal],...]=(
(Decimal('100'),Decimal('2000'),Decimal('0.08')),
(Decimal('2000.01'),Decimal('4000'),Decimal('0.06')),
(Decimal('4000.01'),Decimal('5000'),Decimal('0.04')),
)
@dataclass(slots=True)
class SbpWithdrawalQuote:
usdt_amount: Decimal
rub_amount: Decimal
usdt_exchange_rate: Decimal
service_fee_rate: Decimal
service_fee_usdt: Decimal
class SbpWithdrawalQuoteService:
def __init__(self,*,remote_cache: ICache,logger: ILogger):
self._remote_cache=remote_cache
self._logger=logger
async def _load_rate(self) -> Decimal:
rate_raw=await self._remote_cache.hget('tradex:rub:rate','value')
self._logger.info(f'Actual market value for SBP withdrawal: rate={rate_raw}')
if rate_raw is None:
raise ServiceUnavailableException(message='Exchange rate unavailable')
return Decimal(rate_raw).quantize(Decimal('0.00'),rounding=ROUND_UP)
@staticmethod
def _fee_rate(usdt_amount: Decimal) -> Decimal:
for lo,hi,rate in _FEE_TIERS:
if lo<=usdt_amount<=hi:
return rate
raise OrderTotalOutOfRangeException(message='Withdrawal amount is out of range')
async def get_quote(self,usdt_amount: Decimal) -> SbpWithdrawalQuote:
if usdt_amount<MIN_SBP_WITHDRAWAL_USDT:
raise OrderTotalOutOfRangeException(message='Withdrawal amount is below minimum allowed amount')
if usdt_amount>MAX_SBP_WITHDRAWAL_USDT:
raise OrderTotalOutOfRangeException(message='Withdrawal amount exceeds maximum allowed amount')
rate=await self._load_rate()
fee_rate=self._fee_rate(usdt_amount)
service_fee=(usdt_amount*fee_rate).quantize(Decimal('0.01'))
payout_usdt=(usdt_amount-service_fee).quantize(Decimal('0.01'))
rub_amount=(payout_usdt*rate).quantize(Decimal('1'),rounding=ROUND_DOWN)
return SbpWithdrawalQuote(
usdt_amount=usdt_amount,
rub_amount=rub_amount,
usdt_exchange_rate=rate,
service_fee_rate=fee_rate,
service_fee_usdt=service_fee,
)

33
src/infrastructure/config/settings.py
View File

@@ -90,10 +90,21 @@ class Settings(BaseSettings):
RABBIT_EMAIL_CODE_QUEUE: str = "email.verification_code"
RABBIT_CRYPTO_TRANSFER_QUEUE: str = "crypto.transfer.requested"
RABBIT_CRYPTO_TRANSFER_COMPLETED_QUEUE: str = "crypto.transfer.completed"
RABBIT_SBP_WITHDRAWAL_USDT_REQUESTED_QUEUE: str = 'sbp.withdrawal.usdt.requested'
RABBIT_SBP_WITHDRAWAL_WALLET_EVENTS_QUEUE: str = 'sbp.withdrawal.wallet.events'
ITPAY_PUBLIC_ID: str
ITPAY_API_SECRET: str
MOZEN_API_BASE_URL: str = 'https://extgw.mozen.io'
MOZEN_API_SECRET: str = ''
MOZEN_CLIENT_ID: str = ''
MOZEN_ENDPOINT_ID: str = ''
MOZEN_MERCHANT_ID: str = ''
MOZEN_PARK_ID: str = ''
MOZEN_PARK_TOKEN: str = ''
SBP_WITHDRAWAL_WALLET_ADDRESS: str = ''
CLOUD_KASSIR_PUBLIC_ID: str = ''
CLOUD_KASSIR_API_SECRET: str = ''
@@ -210,6 +221,28 @@ class Settings(BaseSettings):
rb_set('password', 'RABBIT_PASSWORD')
rb_set('vhost', 'RABBIT_VHOST')
extgw = read_secret_optional('extgw')
if extgw:
extgw_ci = {str(k).lower(): v for k, v in extgw.items()}
def extgw_set(field: str, env_key: str) -> None:
v = extgw_ci.get(field)
if v is None:
return
if isinstance(v, str) and not v.strip():
return
data[env_key] = str(v).strip()
extgw_set('api_base_url', 'MOZEN_API_BASE_URL')
extgw_set('api_secret', 'MOZEN_API_SECRET')
extgw_set('client_id', 'MOZEN_CLIENT_ID')
extgw_set('endpoint_id', 'MOZEN_ENDPOINT_ID')
extgw_set('merchant_id', 'MOZEN_MERCHANT_ID')
extgw_set('park_id', 'MOZEN_PARK_ID')
extgw_set('park_token', 'MOZEN_PARK_TOKEN')
extgw_set('wallet_address', 'SBP_WITHDRAWAL_WALLET_ADDRESS')
extgw_set('usdt_wallet_address', 'SBP_WITHDRAWAL_WALLET_ADDRESS')
keydb = read_secret('keydb')
k_ci = {str(k).lower(): v for k, v in keydb.items()}

3
src/infrastructure/database/models/__init__.py
View File

@@ -1,7 +1,8 @@
from src.infrastructure.database.models.base import Base
from src.infrastructure.database.models.order import Order
from src.infrastructure.database.models.payment import Payment
from src.infrastructure.database.models.sbp_withdrawal import SbpWithdrawal
from src.infrastructure.database.models.user import UserModel
__all__ = ['Base','Order','Payment','UserModel']
__all__ = ['Base','Order','Payment','SbpWithdrawal','UserModel']

40
src/infrastructure/database/models/sbp_withdrawal.py Normal file
View File

@@ -0,0 +1,40 @@
from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from sqlalchemy import DateTime,Enum as SAEnum,ForeignKey,Numeric,String,Text
from sqlalchemy.orm import Mapped,mapped_column
from src.application.domain.enums import SbpWithdrawalStatus
from src.infrastructure.database.models.base import Base
from src.infrastructure.database.models.mixins import AuditTimestampsMixin,UlidPrimaryKeyMixin
class SbpWithdrawal(Base,UlidPrimaryKeyMixin,AuditTimestampsMixin):
__tablename__='sbp_withdrawals'
user_id: Mapped[str]=mapped_column(
String(26),
ForeignKey('users.id',ondelete='RESTRICT'),
nullable=False,
index=True,
)
bank_id: Mapped[str]=mapped_column(String(32),nullable=False,index=True)
bank_name: Mapped[str]=mapped_column(String(255),nullable=False)
phone: Mapped[str]=mapped_column(String(32),nullable=False)
wallet_address: Mapped[str]=mapped_column(String(255),nullable=False)
usdt_amount: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False)
rub_amount: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False)
usdt_exchange_rate: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False)
service_fee_rate: Mapped[Decimal]=mapped_column(Numeric(8,4),nullable=False)
service_fee_usdt: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False)
mozen_order_id: Mapped[str | None]=mapped_column(String(128),nullable=True,index=True)
trace_id: Mapped[str]=mapped_column(String(64),nullable=False,index=True)
status: Mapped[SbpWithdrawalStatus]=mapped_column(
SAEnum(SbpWithdrawalStatus,name='sbp_withdrawal_status_enum',values_callable=lambda x:[e.value for e in x]),
nullable=False,
index=True,
default=SbpWithdrawalStatus.CREATED,
)
provider_error: Mapped[str | None]=mapped_column(Text,nullable=True)
usdt_received_at: Mapped[datetime | None]=mapped_column(DateTime(timezone=True),nullable=True)
payout_created_at: Mapped[datetime | None]=mapped_column(DateTime(timezone=True),nullable=True)
payout_completed_at: Mapped[datetime | None]=mapped_column(DateTime(timezone=True),nullable=True)

139
src/infrastructure/database/repositories/sbp_withdrawal_repository.py Normal file
View File

@@ -0,0 +1,139 @@
from __future__ import annotations
from dataclasses import replace
from datetime import datetime,timezone
from sqlalchemy import desc,select
from sqlalchemy.ext.asyncio import AsyncSession
from src.application.abstractions.repositories import ISbpWithdrawalRepository
from src.application.contracts import ILogger
from src.application.domain.entities import SbpWithdrawalEntity
from src.application.domain.enums import SbpWithdrawalStatus
from src.infrastructure.database.models.sbp_withdrawal import SbpWithdrawal
class SbpWithdrawalRepository(ISbpWithdrawalRepository):
def __init__(self,session: AsyncSession,logger: ILogger):
self._session=session
self._logger=logger
@staticmethod
def _to_entity(model: SbpWithdrawal) -> SbpWithdrawalEntity:
return SbpWithdrawalEntity(
id=model.id,
created_at=model.created_at,
updated_at=model.updated_at,
user_id=model.user_id,
bank_id=model.bank_id,
bank_name=model.bank_name,
phone=model.phone,
wallet_address=model.wallet_address,
usdt_amount=model.usdt_amount,
rub_amount=model.rub_amount,
usdt_exchange_rate=model.usdt_exchange_rate,
service_fee_rate=model.service_fee_rate,
service_fee_usdt=model.service_fee_usdt,
mozen_order_id=model.mozen_order_id,
trace_id=model.trace_id,
status=model.status,
provider_error=model.provider_error,
usdt_received_at=model.usdt_received_at,
payout_created_at=model.payout_created_at,
payout_completed_at=model.payout_completed_at,
)
async def create(self,withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalEntity:
values={
'user_id':withdrawal.user_id,
'bank_id':withdrawal.bank_id,
'bank_name':withdrawal.bank_name,
'phone':withdrawal.phone,
'wallet_address':withdrawal.wallet_address,
'usdt_amount':withdrawal.usdt_amount,
'rub_amount':withdrawal.rub_amount,
'usdt_exchange_rate':withdrawal.usdt_exchange_rate,
'service_fee_rate':withdrawal.service_fee_rate,
'service_fee_usdt':withdrawal.service_fee_usdt,
'mozen_order_id':withdrawal.mozen_order_id,
'trace_id':withdrawal.trace_id,
'status':withdrawal.status or SbpWithdrawalStatus.CREATED,
'provider_error':withdrawal.provider_error,
'usdt_received_at':withdrawal.usdt_received_at,
'payout_created_at':withdrawal.payout_created_at,
'payout_completed_at':withdrawal.payout_completed_at,
}
if withdrawal.id is not None:
values['id']=withdrawal.id
model=SbpWithdrawal(**values)
self._session.add(model)
await self._session.flush()
return replace(withdrawal,id=model.id,created_at=model.created_at,updated_at=model.updated_at)
async def get_by_id(self,withdraw_id: str) -> SbpWithdrawalEntity | None:
stmt=select(SbpWithdrawal).where(SbpWithdrawal.id==withdraw_id)
model=await self._session.scalar(stmt)
if model is None:
return None
return self._to_entity(model)
async def get_by_id_for_user(self,*,withdraw_id: str,user_id: str) -> SbpWithdrawalEntity | None:
stmt=select(SbpWithdrawal).where(SbpWithdrawal.id==withdraw_id,SbpWithdrawal.user_id==user_id)
model=await self._session.scalar(stmt)
if model is None:
return None
return self._to_entity(model)
async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[SbpWithdrawalEntity]:
stmt=(
select(SbpWithdrawal)
.where(SbpWithdrawal.user_id==user_id)
.order_by(desc(SbpWithdrawal.created_at))
.limit(limit)
.offset(offset)
)
result=await self._session.scalars(stmt)
return [self._to_entity(model) for model in result.all()]
async def update_status(
self,
*,
withdraw_id: str,
status: SbpWithdrawalStatus,
provider_error: str | None = None,
) -> None:
model=await self._session.get(SbpWithdrawal,withdraw_id)
if model is None:
return
model.status=status
model.provider_error=provider_error
if status==SbpWithdrawalStatus.USDT_RECEIVED:
model.usdt_received_at=datetime.now(timezone.utc)
if status==SbpWithdrawalStatus.PAYOUT_PROCESSING:
model.payout_created_at=datetime.now(timezone.utc)
if status==SbpWithdrawalStatus.PAYOUT_COMPLETED:
model.payout_completed_at=datetime.now(timezone.utc)
await self._session.flush()
async def update_payout_created(
self,
*,
withdraw_id: str,
mozen_order_id: str | None,
status: SbpWithdrawalStatus,
provider_error: str | None,
) -> None:
model=await self._session.get(SbpWithdrawal,withdraw_id)
if model is None:
return
model.mozen_order_id=mozen_order_id
model.status=status
model.provider_error=provider_error
model.payout_created_at=datetime.now(timezone.utc)
if status==SbpWithdrawalStatus.PAYOUT_COMPLETED:
model.payout_completed_at=datetime.now(timezone.utc)
await self._session.flush()

12
src/infrastructure/database/unit_of_work.py
View File

@@ -1,9 +1,10 @@
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from src.application.abstractions import IUnitOfWork
from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,IUserRepository
from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,ISbpWithdrawalRepository,IUserRepository
from src.application.contracts import ILogger
from src.infrastructure.database.repositories.order_repository import OrderRepository
from src.infrastructure.database.repositories.payment_repository import PaymentRepository
from src.infrastructure.database.repositories.sbp_withdrawal_repository import SbpWithdrawalRepository
from src.infrastructure.database.repositories.user_repository import UserRepository
@@ -14,6 +15,7 @@ class UnitOfWork(IUnitOfWork):
self._session: AsyncSession = None
self._order_repository: IOrderRepository | None = None
self._payment_repository: IPaymentRepository | None = None
self._sbp_withdrawal_repository: ISbpWithdrawalRepository | None = None
self._user_repository: IUserRepository | None = None
self._logger: ILogger = logger
@@ -21,6 +23,7 @@ class UnitOfWork(IUnitOfWork):
self._session = self.session_factory()
self._order_repository = None
self._payment_repository = None
self._sbp_withdrawal_repository = None
self._user_repository = None
return self
@@ -49,6 +52,13 @@ class UnitOfWork(IUnitOfWork):
return self._payment_repository
@property
def sbp_withdrawal_repository(self) -> ISbpWithdrawalRepository:
if self._sbp_withdrawal_repository is None:
self._sbp_withdrawal_repository = SbpWithdrawalRepository(session=self._session,logger=self._logger)
return self._sbp_withdrawal_repository
@property
def user_repository(self) -> IUserRepository:
if self._user_repository is None:

4
src/infrastructure/mozen/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
from src.infrastructure.mozen.sbp_client import MozenSbpClient
__all__=['MozenSbpClient']

115
src/infrastructure/mozen/sbp_client.py Normal file
View File

@@ -0,0 +1,115 @@
from __future__ import annotations
from decimal import Decimal,ROUND_DOWN
from typing import Any
import aiohttp
import orjson
from aiohttp import ClientTimeout
from src.application.contracts import IMozenSbpService,SbpBank,SbpPayoutResult
from src.application.domain.exceptions import PaymentProviderException,ServiceUnavailableException
class MozenSbpClient(IMozenSbpService):
def __init__(
self,
*,
api_base_url: str,
client_id: str,
park_token: str,
timeout_seconds: float = 30,
) -> None:
self._api_base_url=api_base_url.rstrip('/')
self._client_id=client_id
self._park_token=park_token
self._timeout=ClientTimeout(total=timeout_seconds)
def _auth_headers(self,trace_id: str) -> dict[str,str]:
if not self._client_id or not self._park_token:
raise ServiceUnavailableException(message='Mozen SBP credentials unavailable')
return {
'Accept':'application/json',
'Content-Type':'application/json',
'x-trace-id':trace_id,
'x-client-id':self._client_id,
'park-token':self._park_token,
}
async def get_banks(self,trace_id: str) -> list[SbpBank]:
url=f'{self._api_base_url}/api/v2/payout/sbp/banks'
headers={'Accept':'application/json','x-trace-id':trace_id}
try:
async with aiohttp.ClientSession(timeout=self._timeout) as session:
async with session.get(url,headers=headers) as resp:
response_text=await resp.text()
try:
response_json=orjson.loads(response_text)
except orjson.JSONDecodeError:
response_json=[]
if resp.status>=400:
raise PaymentProviderException(message='Mozen SBP banks provider error')
if not isinstance(response_json,list):
raise PaymentProviderException(message='Mozen SBP banks response invalid')
banks: list[SbpBank]=[]
for item in response_json:
if not isinstance(item,dict):
continue
bank_id=str(item.get('id') or '').strip()
name=str(item.get('name') or '').strip()
if bank_id and name:
banks.append(SbpBank(id=bank_id,name=name))
return banks
except PaymentProviderException:
raise
except aiohttp.ClientError:
raise PaymentProviderException(message='Mozen SBP banks provider unreachable')
async def make_withdrawal(
self,
*,
user_id: str,
withdraw_id: str,
phone: str,
bank_id: str,
amount: Decimal,
trace_id: str,
) -> SbpPayoutResult:
amount_int=int(amount.quantize(Decimal('1'),rounding=ROUND_DOWN))
payload: dict[str,Any]={
'amount':amount_int,
'order_id':withdraw_id,
'phone':phone,
'bank_id':bank_id,
'description':'SBP withdrawal',
'user_id':user_id,
}
url=f'{self._api_base_url}/api/v2/payout/sbp/withdrawal'
try:
async with aiohttp.ClientSession(timeout=self._timeout) as session:
async with session.post(url,json=payload,headers=self._auth_headers(trace_id)) as resp:
response_text=await resp.text()
try:
response_json=orjson.loads(response_text)
except orjson.JSONDecodeError:
response_json={'raw':response_text}
if resp.status>=400:
raise PaymentProviderException(message='Mozen SBP withdrawal provider error')
if not isinstance(response_json,dict):
raise PaymentProviderException(message='Mozen SBP withdrawal response invalid')
error=response_json.get('error')
if isinstance(error,dict):
error_message=str(error.get('message') or error.get('code') or '').strip() or 'Mozen SBP withdrawal rejected'
return SbpPayoutResult(status=str(response_json.get('status') or 'error'),error=error_message)
request=response_json.get('withdrawal_request')
mozen_order_id=None
if isinstance(request,dict):
mozen_order_id=str(request.get('id') or '').strip() or None
return SbpPayoutResult(
status=str(response_json.get('status') or '').strip(),
mozen_order_id=mozen_order_id,
)
except PaymentProviderException:
raise
except aiohttp.ClientError:
raise PaymentProviderException(message='Mozen SBP withdrawal provider unreachable')

6
src/main.py
View File

@@ -15,9 +15,9 @@ from src.infrastructure.utils import generate_instance_id
from src.infrastructure.logger import logger
from src.infrastructure.config import settings
from src.presentation.handler import application_exception_handler, unhandled_exception_handler
from src.presentation.messaging import crypto_transfer_router
from src.presentation.messaging import crypto_transfer_router,sbp_withdrawal_messaging_router
from src.presentation.middleware import TraceIDMiddleware, SecurityHeadersMiddleware
from src.presentation.routing import order_router,orders_router,payment_router,payments_router
from src.presentation.routing import order_router,orders_router,payment_router,payments_router,sbp_withdrawal_router
security = HTTPBasic()
@@ -78,7 +78,9 @@ app.include_router(order_router)
app.include_router(orders_router)
app.include_router(payment_router)
app.include_router(payments_router)
app.include_router(sbp_withdrawal_router)
app.include_router(crypto_transfer_router)
app.include_router(sbp_withdrawal_messaging_router)
# Added middleware

32
src/presentation/decorators/cache.py
View File

@@ -1,19 +1,20 @@
from __future__ import annotations
import functools
import orjson
from typing import Any,Awaitable,Callable
from fastapi import Request
from fastapi.responses import ORJSONResponse
from src.infrastructure.logger import get_logger
from src.presentation.dependencies.cache import get_cache_remote
def cached(*, prefix: str) -> Callable:
def _to_cache_payload(value: Any) -> Any:
if hasattr(value,'model_dump'):
return value.model_dump(mode='json')
return value
def cached(*,prefix: str,ttl: int = 300,user_scoped: bool = True) -> Callable:
def decorator(func: Callable[..., Awaitable[Any]]):
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
@@ -29,21 +30,34 @@ def cached(*, prefix: str) -> Callable:
auth = kwargs.get('auth')
user_id = getattr(auth, 'user_id', None) if auth else None
if request is None or user_id is None:
if request is None:
return await func(*args, **kwargs)
if user_scoped:
if user_id is None:
return await func(*args, **kwargs)
cache_key = f'{prefix}:{user_id}'
else:
cache_key = prefix
try:
cache = get_cache_remote(request)
hit = await cache.get_user(user_id)
hit = await cache.get(cache_key)
if hit is not None:
logger.debug(f'Cache hit key={cache_key}')
return ORJSONResponse(status_code=200, content=hit)
return ORJSONResponse(status_code=200,content=orjson.loads(hit))
except Exception as e:
logger.warning(f'Cache read failed key={cache_key} error={e}')
return await func(*args, **kwargs)
result = await func(*args, **kwargs)
try:
cache = get_cache_remote(request)
payload = _to_cache_payload(result)
await cache.set(cache_key,orjson.dumps(payload).decode(),ttl)
logger.debug(f'Cache set key={cache_key} ttl={ttl}')
except Exception as e:
logger.warning(f'Cache write failed key={cache_key} error={e}')
return result
return wrapper
return decorator

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

@@ -1,14 +1,15 @@
from __future__ import annotations
from fastapi import Depends
from src.application.abstractions import IUnitOfWork
from src.application.commands import CreateCryptoTransferCompletedCommand,CreateOrderCommand,CreatePaymentCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListOrdersCommand,ListPaymentsCommand
from src.application.contracts import ICache,ILogger,IQueueMessanger,IReceipt
from src.application.commands import CreateCryptoTransferCompletedCommand,CreateOrderCommand,CreatePaymentCommand,CreateSbpWithdrawalCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,GetSbpBanksCommand,GetSbpWithdrawalCommand,HandleSbpWithdrawalWalletEventCommand,ListClientOperationsCommand,ListOrdersCommand,ListPaymentsCommand
from src.application.contracts import ICache,ILogger,IMozenSbpService,IQueueMessanger,IReceipt
from src.application.contracts.i_itpay_service import IItPayService
from src.application.services import PaymentQuoteService
from src.application.services import PaymentQuoteService,SbpWithdrawalQuoteService
from src.infrastructure.cloud_kassir import ClaudeKassirClient
from src.infrastructure.cloud_kassir.constants import CLOUD_KASSIR_API_BASE_URL,CLOUD_KASSIR_FAIL_URL,CLOUD_KASSIR_INN,CLOUD_KASSIR_SUCCESS_URL
from src.infrastructure.config import settings
from src.infrastructure.itpay.client import ItPayClient
from src.infrastructure.mozen import MozenSbpClient
from src.presentation.dependencies.cache import get_remote_cache
from src.presentation.dependencies.logger import get_logger
from src.presentation.dependencies.queue_messanger import get_rabbit
@@ -23,6 +24,14 @@ def get_itpay_service() -> IItPayService:
)
def get_mozen_sbp_service() -> IMozenSbpService:
return MozenSbpClient(
api_base_url=settings.MOZEN_API_BASE_URL,
client_id=settings.MOZEN_CLIENT_ID,
park_token=settings.MOZEN_PARK_TOKEN,
)
def get_create_order_command(
logger: ILogger = Depends(get_logger),
unit_of_work: IUnitOfWork = Depends(get_unit_of_work),
@@ -45,6 +54,43 @@ def get_payment_quote_service(
return PaymentQuoteService(remote_cache=remote_cache,logger=logger)
def get_sbp_withdrawal_quote_service(
logger: ILogger = Depends(get_logger),
remote_cache: ICache = Depends(get_remote_cache),
) -> SbpWithdrawalQuoteService:
return SbpWithdrawalQuoteService(remote_cache=remote_cache,logger=logger)
def get_sbp_banks_command(
logger: ILogger = Depends(get_logger),
mozen_sbp_service: IMozenSbpService = Depends(get_mozen_sbp_service),
) -> GetSbpBanksCommand:
return GetSbpBanksCommand(mozen_sbp_service=mozen_sbp_service,logger=logger)
def get_create_sbp_withdrawal_command(
logger: ILogger = Depends(get_logger),
unit_of_work: IUnitOfWork = Depends(get_unit_of_work),
queue_messanger: IQueueMessanger = Depends(get_rabbit),
mozen_sbp_service: IMozenSbpService = Depends(get_mozen_sbp_service),
quote_service: SbpWithdrawalQuoteService = Depends(get_sbp_withdrawal_quote_service),
) -> CreateSbpWithdrawalCommand:
return CreateSbpWithdrawalCommand(
unit_of_work=unit_of_work,
logger=logger,
queue_messanger=queue_messanger,
mozen_sbp_service=mozen_sbp_service,
quote_service=quote_service,
)
def get_sbp_withdrawal_command(
logger: ILogger = Depends(get_logger),
unit_of_work: IUnitOfWork = Depends(get_unit_of_work),
) -> GetSbpWithdrawalCommand:
return GetSbpWithdrawalCommand(unit_of_work=unit_of_work,logger=logger)
def get_payment_config_command(
logger: ILogger = Depends(get_logger),
payment_quote_service: PaymentQuoteService = Depends(get_payment_quote_service),
@@ -80,6 +126,13 @@ def get_list_payments_command(
return ListPaymentsCommand(unit_of_work=unit_of_work,logger=logger)
def get_list_client_operations_command(
logger: ILogger = Depends(get_logger),
unit_of_work: IUnitOfWork = Depends(get_unit_of_work),
) -> ListClientOperationsCommand:
return ListClientOperationsCommand(unit_of_work=unit_of_work,logger=logger)
def get_payment_command(
logger: ILogger = Depends(get_logger),
unit_of_work: IUnitOfWork = Depends(get_unit_of_work),
@@ -126,3 +179,15 @@ def get_crypto_transfer_completed_command(
receipt: IReceipt = Depends(get_cloud_kassir_receipt),
) -> CreateCryptoTransferCompletedCommand:
return CreateCryptoTransferCompletedCommand(unit_of_work=unit_of_work,receipt=receipt,logger=logger)
def get_sbp_withdrawal_wallet_event_command(
logger: ILogger = Depends(get_logger),
unit_of_work: IUnitOfWork = Depends(get_unit_of_work),
mozen_sbp_service: IMozenSbpService = Depends(get_mozen_sbp_service),
) -> HandleSbpWithdrawalWalletEventCommand:
return HandleSbpWithdrawalWalletEventCommand(
unit_of_work=unit_of_work,
logger=logger,
mozen_sbp_service=mozen_sbp_service,
)

3
src/presentation/messaging/__init__.py
View File

@@ -1,5 +1,6 @@
from src.presentation.messaging.crypto_transfer import crypto_transfer_router
from src.presentation.messaging.sbp_withdrawal import sbp_withdrawal_messaging_router
__all__=['crypto_transfer_router']
__all__=['crypto_transfer_router','sbp_withdrawal_messaging_router']

55
src/presentation/messaging/sbp_withdrawal.py Normal file
View File

@@ -0,0 +1,55 @@
from fastapi import Depends
from faststream.rabbit import RabbitQueue
from faststream.rabbit.fastapi import RabbitMessage,RabbitRouter
from decimal import Decimal
from pydantic import BaseModel
from src.application.commands import HandleSbpWithdrawalWalletEventCommand
from src.application.contracts import ILogger
from src.infrastructure.config import settings
from src.infrastructure.context_vars import trace_id_var
from src.presentation.dependencies.commands import get_sbp_withdrawal_wallet_event_command
from src.presentation.dependencies.logger import get_logger
sbp_withdrawal_messaging_router=RabbitRouter(settings.RABBIT_URL)
class SbpWithdrawalWalletEventMessage(BaseModel):
user_id: str
withdraw_id: str
event_type: str
trace_id: str | None = None
received_usdt_amount: Decimal | None = None
tx_hash: str | None = None
error: str | None = None
@sbp_withdrawal_messaging_router.subscriber(RabbitQueue(settings.RABBIT_SBP_WITHDRAWAL_WALLET_EVENTS_QUEUE,durable=True))
async def sbp_withdrawal_wallet_event_handler(
msg_body: SbpWithdrawalWalletEventMessage,
message: RabbitMessage,
command: HandleSbpWithdrawalWalletEventCommand = Depends(get_sbp_withdrawal_wallet_event_command),
logger: ILogger = Depends(get_logger),
) -> None:
headers=message.headers or {}
trace_id=msg_body.trace_id or message.correlation_id or str(headers.get('trace_id') or '')
token=trace_id_var.set(trace_id)
try:
logger.info({
'event':'sbp_withdrawal_wallet_event_received',
'withdraw_id':msg_body.withdraw_id,
'user_id':msg_body.user_id,
'wallet_event_type':msg_body.event_type,
'rabbit_message_id':message.message_id,
'rabbit_correlation_id':message.correlation_id,
})
await command(
withdraw_id=msg_body.withdraw_id,
user_id=msg_body.user_id,
event_type=msg_body.event_type,
received_usdt_amount=msg_body.received_usdt_amount,
tx_hash=msg_body.tx_hash,
error=msg_body.error,
)
finally:
trace_id_var.reset(token)

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

@@ -1 +1,2 @@
from src.presentation.routing.order import order_router,orders_router,payment_router,payments_router
from src.presentation.routing.sbp_withdrawal import sbp_withdrawal_router

66
src/presentation/routing/order.py
View File

@@ -6,19 +6,19 @@ from fastapi import APIRouter,Depends,Query,Request,WebSocket,WebSocketDisconnec
from fastapi.responses import ORJSONResponse
from fastapi.security.utils import get_authorization_scheme_param
from ulid import ULID
from src.application.commands import CreateOrderCommand,CreatePaymentCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListOrdersCommand,ListPaymentsCommand
from src.application.commands import ClientOperationResult,CreateOrderCommand,CreatePaymentCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListClientOperationsCommand,ListOrdersCommand
from src.application.contracts import IJwtService,ILogger
from src.application.domain.dto import AccessTokenPayload,AuthContext
from src.application.domain.entities import OrderEntity,PaymentEntity
from src.application.domain.entities import OrderEntity,PaymentEntity,SbpWithdrawalEntity
from src.application.domain.enums import OrderStatus
from src.application.domain.exceptions import ApplicationException,ConflictException
from src.application.services import PaymentQuote
from src.infrastructure.context_vars import trace_id_var
from src.presentation.decorators import require_access_token, csrf_protect
from src.presentation.dependencies.commands import get_create_order_command,get_create_payment_command,get_list_orders_command,get_list_payments_command,get_order_command,get_order_status_command,get_payment_command,get_payment_config_command,get_payment_quote_command,get_payment_quote_from_rub_command
from src.presentation.dependencies.commands import get_create_order_command,get_create_payment_command,get_list_client_operations_command,get_list_orders_command,get_order_command,get_order_status_command,get_payment_command,get_payment_config_command,get_payment_quote_command,get_payment_quote_from_rub_command
from src.presentation.dependencies.logger import get_logger
from src.presentation.dependencies.security import get_jwt_service
from src.presentation.schemas.order import CreateOrder,CreateOrderResponse,ErrorResponse,OrderDetailResponse,OrderPaymentResponse,OrdersResponse,OrderStatusResponse,OrderWithPaymentResponse,PaymentConfigResponse,PaymentDetailResponse,PaymentQuoteResponse,PaymentResponse,PaymentsResponse
from src.presentation.schemas.order import ClientOperationResponse,ClientOperationsResponse,CreateOrder,CreateOrderResponse,ErrorResponse,OrderDetailResponse,OrderPaymentResponse,OrdersResponse,OrderStatusResponse,OrderWithPaymentResponse,PaymentConfigResponse,PaymentDetailResponse,PaymentQuoteResponse,PaymentResponse,SbpWithdrawalOperationResponse
from src.presentation.schemas.itpay_payment_models import ItpayPaymentData
order_router = APIRouter(prefix='/order', tags=['orders'])
@@ -110,6 +110,40 @@ def _payment_response(payment: PaymentEntity) -> PaymentResponse:
)
def _withdrawal_operation_response(withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalOperationResponse:
return SbpWithdrawalOperationResponse(
id=withdrawal.id,
created_at=withdrawal.created_at.isoformat() if withdrawal.created_at is not None else None,
updated_at=withdrawal.updated_at.isoformat() if withdrawal.updated_at is not None else None,
user_id=withdrawal.user_id,
bank_id=withdrawal.bank_id,
bank_name=withdrawal.bank_name,
phone=withdrawal.phone,
wallet_address=withdrawal.wallet_address,
usdt_amount=str(withdrawal.usdt_amount) if withdrawal.usdt_amount is not None else None,
rub_amount=str(withdrawal.rub_amount) if withdrawal.rub_amount is not None else None,
usdt_exchange_rate=str(withdrawal.usdt_exchange_rate) if withdrawal.usdt_exchange_rate is not None else None,
service_fee_rate=str(withdrawal.service_fee_rate) if withdrawal.service_fee_rate is not None else None,
service_fee_usdt=str(withdrawal.service_fee_usdt) if withdrawal.service_fee_usdt is not None else None,
mozen_order_id=withdrawal.mozen_order_id,
trace_id=withdrawal.trace_id,
status=withdrawal.status,
provider_error=withdrawal.provider_error,
usdt_received_at=withdrawal.usdt_received_at.isoformat() if withdrawal.usdt_received_at is not None else None,
payout_created_at=withdrawal.payout_created_at.isoformat() if withdrawal.payout_created_at is not None else None,
payout_completed_at=withdrawal.payout_completed_at.isoformat() if withdrawal.payout_completed_at is not None else None,
)
def _client_operation_response(operation: ClientOperationResult) -> ClientOperationResponse:
return ClientOperationResponse(
type=operation.type,
created_at=operation.created_at.isoformat(),
payment=_payment_response(operation.payment) if operation.payment is not None else None,
withdrawal=_withdrawal_operation_response(operation.withdrawal) if operation.withdrawal is not None else None,
)
def _order_detail_response(order: OrderEntity, payment: PaymentEntity | None) -> OrderDetailResponse:
return OrderDetailResponse(
status_code=200,
@@ -266,7 +300,7 @@ async def payment_list_orders(
@payment_router.get(
'/payments',
response_model=PaymentsResponse,
response_model=ClientOperationsResponse,
status_code=200,
responses=ERROR_RESPONSES,
)
@@ -276,12 +310,12 @@ async def payment_list_payments(
limit: int = Query(default=20, ge=1, le=100),
offset: int = Query(default=0, ge=0),
auth: AuthContext = Depends(require_access_token),
command: ListPaymentsCommand = Depends(get_list_payments_command),
) -> PaymentsResponse:
payments = await command(user_id=auth.user_id,limit=limit,offset=offset)
return PaymentsResponse(
command: ListClientOperationsCommand = Depends(get_list_client_operations_command),
) -> ClientOperationsResponse:
operations = await command(user_id=auth.user_id,limit=limit,offset=offset)
return ClientOperationsResponse(
status_code=200,
payments=[_payment_response(payment) for payment in payments],
operations=[_client_operation_response(operation) for operation in operations],
limit=limit,
offset=offset,
)
@@ -325,7 +359,7 @@ async def list_orders(
@payments_router.get(
'/payments',
response_model=PaymentsResponse,
response_model=ClientOperationsResponse,
status_code=200,
responses=ERROR_RESPONSES,
)
@@ -335,12 +369,12 @@ async def list_payments(
limit: int = Query(default=20, ge=1, le=100),
offset: int = Query(default=0, ge=0),
auth: AuthContext = Depends(require_access_token),
command: ListPaymentsCommand = Depends(get_list_payments_command),
) -> PaymentsResponse:
payments = await command(user_id=auth.user_id,limit=limit,offset=offset)
return PaymentsResponse(
command: ListClientOperationsCommand = Depends(get_list_client_operations_command),
) -> ClientOperationsResponse:
operations = await command(user_id=auth.user_id,limit=limit,offset=offset)
return ClientOperationsResponse(
status_code=200,
payments=[_payment_response(payment) for payment in payments],
operations=[_client_operation_response(operation) for operation in operations],
limit=limit,
offset=offset,
)

107
src/presentation/routing/sbp_withdrawal.py Normal file
View File

@@ -0,0 +1,107 @@
from fastapi import APIRouter,Depends,Request
from src.application.commands import CreateSbpWithdrawalCommand,GetSbpBanksCommand,GetSbpWithdrawalCommand
from src.application.domain.dto import AuthContext
from src.application.domain.entities import SbpWithdrawalEntity
from src.presentation.decorators import cached,csrf_protect,require_access_token
from src.presentation.dependencies.commands import get_create_sbp_withdrawal_command,get_sbp_banks_command,get_sbp_withdrawal_command
from src.presentation.schemas.order import ErrorResponse
from src.presentation.schemas.sbp_withdrawal import CreateSbpWithdrawal,CreateSbpWithdrawalResponse,SbpBankResponse,SbpBanksResponse,SbpWithdrawalDetailResponse,SbpWithdrawalResponse
sbp_withdrawal_router=APIRouter(prefix='/sbp-payout',tags=['sbp payouts'])
ERROR_RESPONSES={
400:{'model':ErrorResponse,'description':'Bad Request'},
401:{'model':ErrorResponse,'description':'Unauthorized'},
403:{'model':ErrorResponse,'description':'Forbidden'},
404:{'model':ErrorResponse,'description':'Not Found'},
409:{'model':ErrorResponse,'description':'Conflict'},
422:{'model':ErrorResponse,'description':'Validation Error'},
429:{'model':ErrorResponse,'description':'Too Many Requests'},
500:{'model':ErrorResponse,'description':'Internal Server Error'},
502:{'model':ErrorResponse,'description':'Bad Gateway'},
503:{'model':ErrorResponse,'description':'Service Unavailable'},
}
def _withdrawal_response(withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalResponse:
return SbpWithdrawalResponse(
id=withdrawal.id,
created_at=withdrawal.created_at.isoformat() if withdrawal.created_at is not None else None,
updated_at=withdrawal.updated_at.isoformat() if withdrawal.updated_at is not None else None,
user_id=withdrawal.user_id,
bank_id=withdrawal.bank_id,
bank_name=withdrawal.bank_name,
phone=withdrawal.phone,
wallet_address=withdrawal.wallet_address,
usdt_amount=str(withdrawal.usdt_amount) if withdrawal.usdt_amount is not None else None,
rub_amount=str(withdrawal.rub_amount) if withdrawal.rub_amount is not None else None,
usdt_exchange_rate=str(withdrawal.usdt_exchange_rate) if withdrawal.usdt_exchange_rate is not None else None,
service_fee_rate=str(withdrawal.service_fee_rate) if withdrawal.service_fee_rate is not None else None,
service_fee_usdt=str(withdrawal.service_fee_usdt) if withdrawal.service_fee_usdt is not None else None,
mozen_order_id=withdrawal.mozen_order_id,
trace_id=withdrawal.trace_id,
status=withdrawal.status,
provider_error=withdrawal.provider_error,
usdt_received_at=withdrawal.usdt_received_at.isoformat() if withdrawal.usdt_received_at is not None else None,
payout_created_at=withdrawal.payout_created_at.isoformat() if withdrawal.payout_created_at is not None else None,
payout_completed_at=withdrawal.payout_completed_at.isoformat() if withdrawal.payout_completed_at is not None else None,
)
@sbp_withdrawal_router.get(
'/banks',
response_model=SbpBanksResponse,
status_code=200,
responses=ERROR_RESPONSES,
)
@cached(prefix='sbp:payout:banks',ttl=3600,user_scoped=False)
async def sbp_banks(
request: Request,
command: GetSbpBanksCommand = Depends(get_sbp_banks_command),
) -> SbpBanksResponse:
banks=await command()
return SbpBanksResponse(
status_code=200,
banks=[SbpBankResponse(id=bank.id,name=bank.name) for bank in banks],
)
@sbp_withdrawal_router.post(
'/withdrawals',
response_model=CreateSbpWithdrawalResponse,
status_code=201,
responses=ERROR_RESPONSES,
)
@csrf_protect()
async def create_sbp_withdrawal(
request: Request,
payload: CreateSbpWithdrawal,
auth: AuthContext = Depends(require_access_token),
command: CreateSbpWithdrawalCommand = Depends(get_create_sbp_withdrawal_command),
) -> CreateSbpWithdrawalResponse:
withdrawal=await command(
user_id=auth.user_id,
bank_id=payload.bank_id,
phone=payload.phone,
usdt_amount=payload.usdt_amount,
)
return CreateSbpWithdrawalResponse(status_code=201,withdrawal=_withdrawal_response(withdrawal))
@sbp_withdrawal_router.get(
'/withdrawals/{withdraw_id}',
response_model=SbpWithdrawalDetailResponse,
status_code=200,
responses=ERROR_RESPONSES,
)
@csrf_protect()
async def sbp_withdrawal_detail(
request: Request,
withdraw_id: str,
auth: AuthContext = Depends(require_access_token),
command: GetSbpWithdrawalCommand = Depends(get_sbp_withdrawal_command),
) -> SbpWithdrawalDetailResponse:
withdrawal=await command(withdraw_id=withdraw_id,user_id=auth.user_id)
return SbpWithdrawalDetailResponse(status_code=200,withdrawal=_withdrawal_response(withdrawal))

39
src/presentation/schemas/order.py
View File

@@ -1,6 +1,6 @@
from decimal import Decimal
from pydantic import BaseModel,Field
from src.application.domain.enums import OrderStatus,PaymentStatus
from src.application.domain.enums import OrderStatus,PaymentStatus,SbpWithdrawalStatus
class CreateOrder(BaseModel):
@@ -51,6 +51,36 @@ class PaymentResponse(BaseModel):
expired_date: str | None = None
class SbpWithdrawalOperationResponse(BaseModel):
id: str | None = None
created_at: str | None = None
updated_at: str | None = None
user_id: str | None = None
bank_id: str | None = None
bank_name: str | None = None
phone: str | None = None
wallet_address: str | None = None
usdt_amount: str | None = None
rub_amount: str | None = None
usdt_exchange_rate: str | None = None
service_fee_rate: str | None = None
service_fee_usdt: str | None = None
mozen_order_id: str | None = None
trace_id: str | None = None
status: SbpWithdrawalStatus | None = None
provider_error: str | None = None
usdt_received_at: str | None = None
payout_created_at: str | None = None
payout_completed_at: str | None = None
class ClientOperationResponse(BaseModel):
type: str
created_at: str
payment: PaymentResponse | None = None
withdrawal: SbpWithdrawalOperationResponse | None = None
class CreateOrderResponse(BaseModel):
status_code: int
order: OrderPaymentResponse
@@ -100,6 +130,13 @@ class PaymentsResponse(BaseModel):
offset: int
class ClientOperationsResponse(BaseModel):
status_code: int
operations: list[ClientOperationResponse]
limit: int
offset: int
class PaymentDetailResponse(BaseModel):
status_code: int
payment: PaymentResponse

52
src/presentation/schemas/sbp_withdrawal.py Normal file
View File

@@ -0,0 +1,52 @@
from decimal import Decimal
from pydantic import BaseModel,Field
from src.application.domain.enums import SbpWithdrawalStatus
class SbpBankResponse(BaseModel):
id: str
name: str
class SbpBanksResponse(BaseModel):
status_code: int
banks: list[SbpBankResponse]
class CreateSbpWithdrawal(BaseModel):
bank_id: str = Field(min_length=1,max_length=32)
phone: str = Field(min_length=5,max_length=32)
usdt_amount: Decimal = Field(ge=Decimal('100'),le=Decimal('5000'),decimal_places=2,max_digits=20)
class SbpWithdrawalResponse(BaseModel):
id: str | None = None
created_at: str | None = None
updated_at: str | None = None
user_id: str | None = None
bank_id: str | None = None
bank_name: str | None = None
phone: str | None = None
wallet_address: str | None = None
usdt_amount: str | None = None
rub_amount: str | None = None
usdt_exchange_rate: str | None = None
service_fee_rate: str | None = None
service_fee_usdt: str | None = None
mozen_order_id: str | None = None
trace_id: str | None = None
status: SbpWithdrawalStatus | None = None
provider_error: str | None = None
usdt_received_at: str | None = None
payout_created_at: str | None = None
payout_completed_at: str | None = None
class CreateSbpWithdrawalResponse(BaseModel):
status_code: int
withdrawal: SbpWithdrawalResponse
class SbpWithdrawalDetailResponse(BaseModel):
status_code: int
withdrawal: SbpWithdrawalResponse