k1nq
41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
import logging
|
|
|
|
from faststream import FastStream
|
|
from faststream.exceptions import RejectMessage
|
|
from faststream.rabbit import RabbitBroker
|
|
|
|
from app.config import settings
|
|
from app.db import close_database, session_factory
|
|
from app.messaging import PAYMENTS_EXCHANGE, PAYMENTS_QUEUE, declare_topology
|
|
from app.processing import PaymentGateway, PaymentProcessor, WebhookClient
|
|
from app.schemas import PaymentCreatedEvent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
broker = RabbitBroker(settings.rabbitmq_url, graceful_timeout=30.0)
|
|
app = FastStream(broker)
|
|
processor = PaymentProcessor(
|
|
session_factory,
|
|
PaymentGateway(settings),
|
|
WebhookClient(settings.webhook_timeout),
|
|
settings,
|
|
)
|
|
|
|
|
|
@broker.subscriber(PAYMENTS_QUEUE, PAYMENTS_EXCHANGE)
|
|
async def handle_payment(event: PaymentCreatedEvent) -> None:
|
|
try:
|
|
await processor.process(event)
|
|
except Exception as exc:
|
|
logger.exception("Payment %s exhausted all processing attempts", event.payment_id)
|
|
raise RejectMessage() from exc
|
|
|
|
|
|
@app.after_startup
|
|
async def setup_topology() -> None:
|
|
await declare_topology(broker)
|
|
|
|
|
|
@app.after_shutdown
|
|
async def shutdown_database() -> None:
|
|
await close_database()
|