40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
from fastapi import Depends
|
|
import orjson
|
|
from faststream.rabbit.fastapi import RabbitMessage,RabbitRouter
|
|
from pydantic import BaseModel
|
|
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.logger import get_logger
|
|
|
|
|
|
crypto_transfer_router=RabbitRouter(settings.RABBIT_URL)
|
|
|
|
|
|
class CryptoTransferCompletedMessage(BaseModel):
|
|
user_id: str
|
|
order_id: str
|
|
trace_id: str
|
|
message_id: str
|
|
|
|
|
|
@crypto_transfer_router.subscriber(settings.RABBIT_CRYPTO_TRANSFER_COMPLETED_QUEUE)
|
|
async def crypto_transfer_completed_handler(
|
|
msg_body: CryptoTransferCompletedMessage,
|
|
message: RabbitMessage,
|
|
logger: ILogger = Depends(get_logger),
|
|
) -> None:
|
|
trace_id=msg_body.trace_id
|
|
token=trace_id_var.set(trace_id)
|
|
try:
|
|
payload=msg_body.model_dump(mode='json')
|
|
logger.info(orjson.dumps({
|
|
'event':'crypto_transfer_completed_received',
|
|
'payload':payload,
|
|
'rabbit_message_id':message.message_id,
|
|
'rabbit_correlation_id':message.correlation_id,
|
|
},default=str).decode())
|
|
finally:
|
|
trace_id_var.reset(token)
|
|
|