k1nq
58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
import asyncio
|
|
import logging
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI
|
|
from faststream.rabbit import Channel, RabbitBroker
|
|
|
|
from app.api import router
|
|
from app.config import settings
|
|
from app.db import close_database, session_factory
|
|
from app.messaging import declare_topology
|
|
from app.outbox import OutboxPublisher
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
|
|
broker = RabbitBroker(
|
|
settings.rabbitmq_url,
|
|
default_channel=Channel(publisher_confirms=True, on_return_raises=True),
|
|
)
|
|
await broker.start()
|
|
await declare_topology(broker)
|
|
|
|
stop_event = asyncio.Event()
|
|
publisher = OutboxPublisher(
|
|
session_factory,
|
|
broker,
|
|
batch_size=settings.outbox_batch_size,
|
|
poll_interval=settings.outbox_poll_interval,
|
|
publish_timeout=settings.outbox_publish_timeout,
|
|
)
|
|
publisher_task = asyncio.create_task(publisher.run(stop_event))
|
|
|
|
try:
|
|
yield
|
|
finally:
|
|
stop_event.set()
|
|
try:
|
|
await asyncio.wait_for(
|
|
publisher_task,
|
|
timeout=settings.outbox_publish_timeout + settings.outbox_poll_interval,
|
|
)
|
|
except TimeoutError:
|
|
publisher_task.cancel()
|
|
await asyncio.gather(publisher_task, return_exceptions=True)
|
|
await broker.stop()
|
|
await close_database()
|
|
|
|
|
|
app = FastAPI(title=settings.app_name, version="1.0.0", lifespan=lifespan)
|
|
app.include_router(router)
|