acemq_fastapi
FastAPI integration for AceMQ.
A lifespan that owns the connection and the consumers, dependencies for publishing from a route, a decorator for registering handlers, and a health check that knows a blocked broker is not a reason to restart.
::
from acemq_amqp import Ack, Message, accept
from acemq_fastapi import AceMQ, publishes
from fastapi import Depends, FastAPI
acemq = AceMQ()
@acemq.consumer("orders")
async def handle(message: Message) -> Ack:
...
return accept()
app = FastAPI(lifespan=acemq.lifespan)
app.include_router(acemq.health_router())
@app.post("/orders")
async def place(orders=Depends(publishes("orders"))) -> dict[str, str]:
result = await orders.send({"id": "1"})
return {"id": result.message_id}
This package re-exports nothing from acemq_amqp. Messages, acknowledgements,
codecs and retry policies come from the library under their own names, so the two
imports say which half of the system a line is talking about.
1# Copyright 2026 AceMQ. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# https://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15"""FastAPI integration for AceMQ. 16 17A lifespan that owns the connection and the consumers, dependencies for 18publishing from a route, a decorator for registering handlers, and a health check 19that knows a blocked broker is not a reason to restart. 20 21:: 22 23 from acemq_amqp import Ack, Message, accept 24 from acemq_fastapi import AceMQ, publishes 25 from fastapi import Depends, FastAPI 26 27 acemq = AceMQ() 28 29 @acemq.consumer("orders") 30 async def handle(message: Message) -> Ack: 31 ... 32 return accept() 33 34 app = FastAPI(lifespan=acemq.lifespan) 35 app.include_router(acemq.health_router()) 36 37 @app.post("/orders") 38 async def place(orders=Depends(publishes("orders"))) -> dict[str, str]: 39 result = await orders.send({"id": "1"}) 40 return {"id": result.message_id} 41 42This package re-exports nothing from :mod:`acemq_amqp`. Messages, acknowledgements, 43codecs and retry policies come from the library under their own names, so the two 44imports say which half of the system a line is talking about. 45""" 46 47from .dependencies import ( 48 AceMQConnection, 49 AceMQDep, 50 get_acemq, 51 get_connection, 52 publishes, 53) 54from .health import AceMQHealth, health_router, report_as_json 55from .integration import AceMQ, ConsumerRegistration, DrainReport 56from .lifespans import Lifespan, compose_lifespans 57from .settings import ( 58 AceMQSettings, 59 BindingSettings, 60 ConsumerSettings, 61 ExchangeSettings, 62 HealthSettings, 63 QueueSettings, 64 RetrySettings, 65 TlsSettings, 66 TopologySettings, 67) 68 69__version__ = "0.1.5" 70 71__all__ = [ 72 "AceMQ", 73 "AceMQConnection", 74 "AceMQDep", 75 "AceMQHealth", 76 "AceMQSettings", 77 "BindingSettings", 78 "ConsumerRegistration", 79 "ConsumerSettings", 80 "DrainReport", 81 "ExchangeSettings", 82 "HealthSettings", 83 "Lifespan", 84 "QueueSettings", 85 "RetrySettings", 86 "TlsSettings", 87 "TopologySettings", 88 "__version__", 89 "compose_lifespans", 90 "get_acemq", 91 "get_connection", 92 "health_router", 93 "publishes", 94 "report_as_json", 95]
class
AceMQ:
122class AceMQ: 123 """The connection, the consumers and the settings, for one application. 124 125 :: 126 127 acemq = AceMQ() 128 129 @acemq.consumer("orders") 130 async def handle(message: Message) -> Ack: 131 ... 132 return accept() 133 134 app = FastAPI(lifespan=acemq.lifespan) 135 136 :param settings: what to connect to and how. Read from the environment when 137 not given 138 :param connection_factory: what opens the connection. Replaceable for tests, 139 and for an application that builds its own 140 :param on_publish: interceptors wrapping every publish, outermost first 141 :param on_consume: interceptors wrapping every handler 142 :param observer: where the library's numbers go 143 """ 144 145 def __init__( 146 self, 147 settings: AceMQSettings | None = None, 148 *, 149 connection_factory: ConnectionFactory | None = None, 150 on_publish: tuple[PublishInterceptor, ...] = (), 151 on_consume: tuple[ConsumeInterceptor, ...] = (), 152 observer: Observer | None = None, 153 ) -> None: 154 self._settings = settings if settings is not None else AceMQSettings() 155 self._connect: ConnectionFactory = connection_factory or self._open 156 self._on_publish = tuple(on_publish) 157 self._on_consume = tuple(on_consume) 158 self._observer = observer 159 self._registrations: dict[str, ConsumerRegistration] = {} 160 self._consumers: dict[str, Consumer] = {} 161 self._publishers: dict[tuple[Any, ...], Publisher] = {} 162 self._connection: Connection | None = None 163 self._lock = asyncio.Lock() 164 165 # ------------------------------------------------------------------ state 166 167 @property 168 def settings(self) -> AceMQSettings: 169 """What this integration was configured with.""" 170 return self._settings 171 172 @property 173 def started(self) -> bool: 174 """Whether the connection is open.""" 175 return self._connection is not None 176 177 @property 178 def connection(self) -> Connection: 179 """The open connection. 180 181 :raises AceMQError: before the lifespan has opened it, because the 182 alternative is a ``None`` that turns into an ``AttributeError`` 183 three frames away from the mistake 184 """ 185 if self._connection is None: 186 raise AceMQError( 187 "acemq-fastapi: there is no connection yet. It is opened by the lifespan, " 188 "so this is either a call made before the application started or an " 189 "application built without lifespan=acemq.lifespan" 190 ) 191 return self._connection 192 193 @property 194 def registrations(self) -> tuple[ConsumerRegistration, ...]: 195 """Every ``@acemq.consumer`` declaration, in the order they were made.""" 196 return tuple(self._registrations.values()) 197 198 @property 199 def consumers(self) -> Mapping[str, Consumer]: 200 """The running consumers by name, for scaling, pausing and counters.""" 201 return dict(self._consumers) 202 203 # -------------------------------------------------------------- consumers 204 205 def consumer( 206 self, 207 queue: str, 208 *, 209 name: str | None = None, 210 codec: Codec | None = None, 211 retry: RetryPolicy | None = None, 212 prefetch: int | None = None, 213 concurrency: int | None = None, 214 declare: bool | None = None, 215 tag: str = "", 216 args: Mapping[str, Any] | None = None, 217 ) -> Callable[[HandlerT], HandlerT]: 218 """Registers a handler for a queue. The Python analogue of ``@AceListener``. 219 220 The decorated function is returned unchanged, so it stays an ordinary 221 function that a test can call with a :class:`~acemq_amqp.Message` and 222 assert on, without a broker anywhere near it. 223 224 Nothing subscribes here. The registration is held until the lifespan 225 opens, which is what puts the handler on the application's own event loop 226 rather than on whichever loop happened to import the module. 227 228 :param queue: what to read 229 :param name: what to call this consumer in :attr:`consumers`. Defaults to 230 the queue 231 :param codec: a codec other than the connection's 232 :param retry: a policy other than ``acemq.consumer.retry`` 233 :param prefetch: how many unacknowledged messages to hold 234 :param concurrency: how many messages to work on at once 235 :param declare: declare the queues a failure needs before subscribing 236 :param tag: what to call this consumer to the broker 237 :param args: broker-specific consumer arguments 238 :returns: the decorator 239 :raises AceMQError: when the name is already taken, or the lifespan has 240 already opened 241 """ 242 243 def register(handler: HandlerT) -> HandlerT: 244 self.add_consumer( 245 queue, 246 handler, 247 name=name, 248 codec=codec, 249 retry=retry, 250 prefetch=prefetch, 251 concurrency=concurrency, 252 declare=declare, 253 tag=tag, 254 args=args, 255 ) 256 return handler 257 258 return register 259 260 def add_consumer( 261 self, 262 queue: str, 263 handler: Handler, 264 *, 265 name: str | None = None, 266 codec: Codec | None = None, 267 retry: RetryPolicy | None = None, 268 prefetch: int | None = None, 269 concurrency: int | None = None, 270 declare: bool | None = None, 271 tag: str = "", 272 args: Mapping[str, Any] | None = None, 273 ) -> ConsumerRegistration: 274 """Registers a handler without the decorator. 275 276 What a factory function or a plugin uses. Same arguments as 277 :meth:`consumer`, and the same restriction: before the lifespan opens. 278 """ 279 if self._connection is not None: 280 raise AceMQError( 281 f"acemq-fastapi: cannot register a consumer for {queue!r} now — the " 282 "lifespan has already opened. Register consumers while the module is " 283 "being imported, which is what the decorator is for" 284 ) 285 chosen = name or queue 286 if chosen in self._registrations: 287 raise AceMQError( 288 f"acemq-fastapi: there is already a consumer called {chosen!r}. Two " 289 "consumers on one queue is a reasonable thing to want; give one of them " 290 "a name=... so they can be told apart" 291 ) 292 registration = ConsumerRegistration( 293 name=chosen, 294 queue=queue, 295 handler=handler, 296 codec=codec, 297 retry=retry, 298 prefetch=prefetch, 299 concurrency=concurrency, 300 declare=declare, 301 tag=tag, 302 args=dict(args or {}), 303 ) 304 self._registrations[chosen] = registration 305 return registration 306 307 async def start_consumer(self, name: str) -> Consumer: 308 """Subscribes one registered consumer that is not running. 309 310 What :attr:`consumers` is for in the other direction: a consumer stopped 311 for back pressure, or one left registered by 312 ``acemq.consumer.auto_start=false``, is started again by name. 313 """ 314 registration = self._registrations.get(name) 315 if registration is None: 316 known = ", ".join(sorted(self._registrations)) or "none" 317 raise AceMQError( 318 f"acemq-fastapi: no consumer is registered as {name!r}; registered: {known}" 319 ) 320 if name in self._consumers: 321 return self._consumers[name] 322 consumer = await self._subscribe(registration) 323 self._consumers[name] = consumer 324 return consumer 325 326 async def stop_consumer(self, name: str) -> None: 327 """Drains and closes one consumer, leaving it registered. 328 329 It finishes what its handlers are holding, exactly as a shutdown does. 330 """ 331 consumer = self._consumers.pop(name, None) 332 if consumer is not None: 333 await consumer.close() 334 335 # ------------------------------------------------------------- publishing 336 337 def publisher( 338 self, 339 exchange: str = "", 340 routing_key: str = "", 341 *, 342 codec: Codec | None = None, 343 persistent: bool = True, 344 mandatory: bool = False, 345 ) -> Publisher: 346 """A publisher for a destination, built once and reused. 347 348 Publishers are cheap to keep and safe to share, and building one per 349 request is a small waste repeated a lot. This caches by every argument 350 that changes what a publisher does, so two routes asking for the same 351 destination get the same object and a route asking for a different one 352 does not quietly get somebody else's. 353 354 :raises AceMQError: before the lifespan has opened the connection 355 """ 356 key = (exchange, routing_key, id(codec) if codec else None, persistent, mandatory) 357 publisher = self._publishers.get(key) 358 if publisher is None: 359 publisher = self.connection.publisher( 360 exchange, 361 routing_key, 362 codec=codec, 363 persistent=persistent, 364 mandatory=mandatory, 365 ) 366 self._publishers[key] = publisher 367 return publisher 368 369 # ----------------------------------------------------------------- health 370 371 def health_check(self) -> AceMQHealth: 372 """This connection as a :class:`~acemq_amqp.HealthCheck`. 373 374 Pass it to :func:`acemq_amqp.aggregate_health` beside an application's 375 own checks, or mount :meth:`health_router` and let it answer on its own. 376 """ 377 from .health import AceMQHealth 378 379 return AceMQHealth(self) 380 381 async def health(self) -> acemq_amqp.HealthReport: 382 """What a readiness probe should see. See :class:`~acemq_fastapi.AceMQHealth`.""" 383 return await self.health_check().check() 384 385 def health_router(self, *, path: str | None = None, tags: list[Any] | None = None) -> Any: 386 """An ``APIRouter`` with one route reporting on the connection. 387 388 ``app.include_router(acemq.health_router())`` and nothing else. Mounting 389 it is explicit rather than automatic because a route that appears in an 390 application's OpenAPI document without anybody adding it is a route 391 somebody has to go looking for. 392 """ 393 from .health import health_router 394 395 return health_router(self, path=path, tags=tags) 396 397 # --------------------------------------------------------------- lifespan 398 399 @asynccontextmanager 400 async def lifespan(self, app: FastAPI | None = None) -> AsyncIterator[Mapping[str, Any]]: 401 """Opens the connection and the consumers, and drains them on the way out. 402 403 ``FastAPI(lifespan=acemq.lifespan)``. To keep an application's own 404 lifespan as well, see :func:`acemq_fastapi.compose_lifespans` — this one 405 belongs innermost, so the consumers start after whatever they depend on 406 and drain before it goes away. 407 408 The ``app`` is optional so the same context manager can be used in a test 409 or a worker with no ASGI application at all. When one is given it gets 410 ``app.state.acemq``, which is how the request dependencies find this 411 object without a module-level global. 412 """ 413 started = await self.start(app) 414 try: 415 yield {"acemq": self} 416 finally: 417 report = await self.drain() 418 if not report.finished: 419 log.warning( 420 "acemq-fastapi: the drain did not finish in %s; %d message(s) were " 421 "left unsettled and the broker will redeliver them", 422 self._settings.consumer.shutdown_timeout, 423 report.abandoned, 424 ) 425 await self.aclose() 426 log.info( 427 "acemq-fastapi: closed after %.3fs, %d consumer(s) drained", 428 report.waited, 429 started, 430 ) 431 432 async def start(self, app: FastAPI | None = None) -> int: 433 """Connects, applies the topology and subscribes the consumers. 434 435 Called by :meth:`lifespan`. Public because a worker process that serves no 436 HTTP at all still wants the same start-up, and should not have to 437 construct an ASGI application to get it. 438 439 :returns: how many consumers were started 440 """ 441 async with self._lock: 442 if self._connection is not None: 443 raise AceMQError("acemq-fastapi: this integration is already started") 444 timeout = self._settings.connection_timeout.total_seconds() 445 try: 446 connection = await asyncio.wait_for(self._connect(self._settings), timeout) 447 except asyncio.TimeoutError as expired: 448 raise AceMQError( 449 f"acemq-fastapi: the broker at {self._settings.broker_url} did not " 450 f"answer within {timeout:g}s" 451 ) from expired 452 self._connection = connection 453 454 try: 455 for publishing in self._on_publish: 456 connection.intercept_publish(publishing) 457 for consuming in self._on_consume: 458 connection.intercept_consume(consuming) 459 460 topology = self._settings.build_topology() 461 if topology is not None: 462 await connection.declare(topology) 463 464 if self._settings.consumer.auto_start: 465 for registration in self._registrations.values(): 466 self._consumers[registration.name] = await self._subscribe(registration) 467 except BaseException: 468 # A half-started application is worse than one that refused to start: 469 # the routes would be up, the consumers would not, and nothing outside 470 # could tell. Undo what was done and let the failure out. 471 await self.drain() 472 await self.aclose() 473 raise 474 475 if app is not None: 476 app.state.acemq = self 477 log.info( 478 "acemq-fastapi: connected to %s, %d consumer(s) running", 479 self._settings.broker_url, 480 len(self._consumers), 481 ) 482 return len(self._consumers) 483 484 async def drain(self) -> DrainReport: 485 """Stops every consumer, finishing the handlers already running. 486 487 What this finishes, and what it gives back, is written down in the README 488 under *What a drain finishes*. In short: a handler in flight runs to 489 completion; a delivery fetched but not yet started is handed back to the 490 broker; a retry waiting out a short backoff in this process is waited out 491 in full; a publish waiting for its confirm is not waited for. 492 493 Bounded by ``acemq.consumer.shutdown_timeout``. When that expires the 494 handlers still running are cancelled and their messages left unsettled — 495 the broker redelivers them, so the work may be done twice and nothing is 496 dropped. 497 """ 498 if self._connection is None or not self._consumers: 499 self._consumers.clear() 500 return DrainReport(finished=True, waited=0.0, consumers=0) 501 502 consumers = tuple(self._consumers.values()) 503 self._consumers.clear() 504 deadline = self._settings.consumer.shutdown_timeout.total_seconds() 505 loop = asyncio.get_running_loop() 506 started = loop.time() 507 508 closing = [asyncio.create_task(consumer.close()) for consumer in consumers] 509 done, pending = await asyncio.wait(closing, timeout=deadline) 510 waited = loop.time() - started 511 512 if not pending: 513 for task in done: 514 if task.cancelled(): 515 # Its workers were already gone — cancelled from outside, or 516 # killed by something the library logged. Nothing here can 517 # finish a handler that is not running any more. 518 log.warning( 519 "acemq-fastapi: a consumer's workers had already stopped; " 520 "whatever they were holding is back with the broker" 521 ) 522 elif task.exception() is not None: 523 log.warning( 524 "acemq-fastapi: a consumer failed to close: %s", task.exception() 525 ) 526 return DrainReport(finished=True, waited=waited, consumers=len(consumers)) 527 528 # Out of time. Cancelling is the only lever there is, and it is pulled 529 # last rather than first: it aborts the handlers still running and leaves 530 # their deliveries unsettled for the broker to hand on. That is the price 531 # of a deadline, and it is cheaper than being SIGKILLed with the same 532 # messages unsettled and the socket gone. 533 abandoned = sum(consumer.in_flight for consumer in consumers) 534 for task in pending: 535 task.cancel() 536 await asyncio.gather(*pending, return_exceptions=True) 537 return DrainReport( 538 finished=False, 539 waited=loop.time() - started, 540 consumers=len(consumers), 541 abandoned=abandoned, 542 ) 543 544 async def aclose(self) -> None: 545 """Releases the connection. Idempotent, and safe after a failed start.""" 546 connection = self._connection 547 self._connection = None 548 self._publishers.clear() 549 self._consumers.clear() 550 if connection is not None: 551 with suppress(Exception): 552 await connection.close() 553 554 # ---------------------------------------------------------------- private 555 556 async def _subscribe(self, registration: ConsumerRegistration) -> Consumer: 557 defaults = self._settings.consumer 558 return await self.connection.consume( 559 registration.queue, 560 registration.handler, 561 codec=registration.codec, 562 retry=registration.retry or self._settings.build_retry(), 563 prefetch=registration.prefetch 564 if registration.prefetch is not None 565 else defaults.prefetch, 566 concurrency=registration.concurrency 567 if registration.concurrency is not None 568 else defaults.concurrency, 569 tag=registration.tag, 570 args=registration.args, 571 declare=( 572 registration.declare if registration.declare is not None else defaults.declare 573 ), 574 ) 575 576 async def _open(self, settings: AceMQSettings) -> Connection: 577 """The default factory: :func:`acemq_amqp.connect` with the settings applied.""" 578 return await acemq_amqp.connect( 579 settings.broker_url, 580 codec=settings.build_codec(), 581 origin=settings.client_name, 582 retry=settings.build_retry(), 583 prefetch=settings.prefetch, 584 max_outstanding_publishes=settings.max_outstanding_publishes, 585 confirm_timeout=settings.confirm_timeout, 586 security=settings.build_security(), 587 observer=self._observer, 588 )
The connection, the consumers and the settings, for one application.
::
acemq = AceMQ()
@acemq.consumer("orders")
async def handle(message: Message) -> Ack:
...
return accept()
app = FastAPI(lifespan=acemq.lifespan)
Parameters
- settings: what to connect to and how. Read from the environment when not given
- connection_factory: what opens the connection. Replaceable for tests, and for an application that builds its own
- on_publish: interceptors wrapping every publish, outermost first
- on_consume: interceptors wrapping every handler
- observer: where the library's numbers go
AceMQ( settings: AceMQSettings | None = None, *, connection_factory: Callable[[AceMQSettings], Awaitable[acemq_amqp.connection.Connection]] | None = None, on_publish: tuple[Callable[[acemq_amqp.interceptors.PublishContext, Callable[[acemq_amqp.interceptors.PublishContext], Awaitable[acemq_amqp.transport.PublishResult]]], Awaitable[acemq_amqp.transport.PublishResult]], ...] = (), on_consume: tuple[Callable[[acemq_amqp.interceptors.ConsumeContext, Callable[[acemq_amqp.interceptors.ConsumeContext], Awaitable[acemq_amqp.ack.Ack]]], Awaitable[acemq_amqp.ack.Ack]], ...] = (), observer: acemq_amqp.telemetry.Observer | None = None)
145 def __init__( 146 self, 147 settings: AceMQSettings | None = None, 148 *, 149 connection_factory: ConnectionFactory | None = None, 150 on_publish: tuple[PublishInterceptor, ...] = (), 151 on_consume: tuple[ConsumeInterceptor, ...] = (), 152 observer: Observer | None = None, 153 ) -> None: 154 self._settings = settings if settings is not None else AceMQSettings() 155 self._connect: ConnectionFactory = connection_factory or self._open 156 self._on_publish = tuple(on_publish) 157 self._on_consume = tuple(on_consume) 158 self._observer = observer 159 self._registrations: dict[str, ConsumerRegistration] = {} 160 self._consumers: dict[str, Consumer] = {} 161 self._publishers: dict[tuple[Any, ...], Publisher] = {} 162 self._connection: Connection | None = None 163 self._lock = asyncio.Lock()
settings: AceMQSettings
167 @property 168 def settings(self) -> AceMQSettings: 169 """What this integration was configured with.""" 170 return self._settings
What this integration was configured with.
started: bool
172 @property 173 def started(self) -> bool: 174 """Whether the connection is open.""" 175 return self._connection is not None
Whether the connection is open.
connection: acemq_amqp.connection.Connection
177 @property 178 def connection(self) -> Connection: 179 """The open connection. 180 181 :raises AceMQError: before the lifespan has opened it, because the 182 alternative is a ``None`` that turns into an ``AttributeError`` 183 three frames away from the mistake 184 """ 185 if self._connection is None: 186 raise AceMQError( 187 "acemq-fastapi: there is no connection yet. It is opened by the lifespan, " 188 "so this is either a call made before the application started or an " 189 "application built without lifespan=acemq.lifespan" 190 ) 191 return self._connection
The open connection.
Raises
- AceMQError: before the lifespan has opened it, because the
alternative is a
Nonethat turns into anAttributeErrorthree frames away from the mistake
registrations: tuple[ConsumerRegistration, ...]
193 @property 194 def registrations(self) -> tuple[ConsumerRegistration, ...]: 195 """Every ``@acemq.consumer`` declaration, in the order they were made.""" 196 return tuple(self._registrations.values())
Every @acemq.consumer declaration, in the order they were made.
consumers: Mapping[str, acemq_amqp.connection.Consumer]
198 @property 199 def consumers(self) -> Mapping[str, Consumer]: 200 """The running consumers by name, for scaling, pausing and counters.""" 201 return dict(self._consumers)
The running consumers by name, for scaling, pausing and counters.
def
consumer( self, queue: str, *, name: str | None = None, codec: acemq_amqp.codec.Codec | None = None, retry: acemq_amqp.retry.RetryPolicy | None = None, prefetch: int | None = None, concurrency: int | None = None, declare: bool | None = None, tag: str = '', args: Mapping[str, typing.Any] | None = None) -> Callable[[~HandlerT], ~HandlerT]:
205 def consumer( 206 self, 207 queue: str, 208 *, 209 name: str | None = None, 210 codec: Codec | None = None, 211 retry: RetryPolicy | None = None, 212 prefetch: int | None = None, 213 concurrency: int | None = None, 214 declare: bool | None = None, 215 tag: str = "", 216 args: Mapping[str, Any] | None = None, 217 ) -> Callable[[HandlerT], HandlerT]: 218 """Registers a handler for a queue. The Python analogue of ``@AceListener``. 219 220 The decorated function is returned unchanged, so it stays an ordinary 221 function that a test can call with a :class:`~acemq_amqp.Message` and 222 assert on, without a broker anywhere near it. 223 224 Nothing subscribes here. The registration is held until the lifespan 225 opens, which is what puts the handler on the application's own event loop 226 rather than on whichever loop happened to import the module. 227 228 :param queue: what to read 229 :param name: what to call this consumer in :attr:`consumers`. Defaults to 230 the queue 231 :param codec: a codec other than the connection's 232 :param retry: a policy other than ``acemq.consumer.retry`` 233 :param prefetch: how many unacknowledged messages to hold 234 :param concurrency: how many messages to work on at once 235 :param declare: declare the queues a failure needs before subscribing 236 :param tag: what to call this consumer to the broker 237 :param args: broker-specific consumer arguments 238 :returns: the decorator 239 :raises AceMQError: when the name is already taken, or the lifespan has 240 already opened 241 """ 242 243 def register(handler: HandlerT) -> HandlerT: 244 self.add_consumer( 245 queue, 246 handler, 247 name=name, 248 codec=codec, 249 retry=retry, 250 prefetch=prefetch, 251 concurrency=concurrency, 252 declare=declare, 253 tag=tag, 254 args=args, 255 ) 256 return handler 257 258 return register
Registers a handler for a queue. The Python analogue of @AceListener.
The decorated function is returned unchanged, so it stays an ordinary
function that a test can call with a ~acemq_amqp.Message and
assert on, without a broker anywhere near it.
Nothing subscribes here. The registration is held until the lifespan opens, which is what puts the handler on the application's own event loop rather than on whichever loop happened to import the module.
Parameters
- queue: what to read
- name: what to call this consumer in
consumers. Defaults to the queue - codec: a codec other than the connection's
- retry: a policy other than
acemq.consumer.retry - prefetch: how many unacknowledged messages to hold
- concurrency: how many messages to work on at once
- declare: declare the queues a failure needs before subscribing
- tag: what to call this consumer to the broker
- args: broker-specific consumer arguments :returns: the decorator
Raises
- AceMQError: when the name is already taken, or the lifespan has already opened
def
add_consumer( self, queue: str, handler: Callable[acemq_amqp.connection.Message, acemq_amqp.ack.Ack | Awaitable[acemq_amqp.ack.Ack]], *, name: str | None = None, codec: acemq_amqp.codec.Codec | None = None, retry: acemq_amqp.retry.RetryPolicy | None = None, prefetch: int | None = None, concurrency: int | None = None, declare: bool | None = None, tag: str = '', args: Mapping[str, typing.Any] | None = None) -> ConsumerRegistration:
260 def add_consumer( 261 self, 262 queue: str, 263 handler: Handler, 264 *, 265 name: str | None = None, 266 codec: Codec | None = None, 267 retry: RetryPolicy | None = None, 268 prefetch: int | None = None, 269 concurrency: int | None = None, 270 declare: bool | None = None, 271 tag: str = "", 272 args: Mapping[str, Any] | None = None, 273 ) -> ConsumerRegistration: 274 """Registers a handler without the decorator. 275 276 What a factory function or a plugin uses. Same arguments as 277 :meth:`consumer`, and the same restriction: before the lifespan opens. 278 """ 279 if self._connection is not None: 280 raise AceMQError( 281 f"acemq-fastapi: cannot register a consumer for {queue!r} now — the " 282 "lifespan has already opened. Register consumers while the module is " 283 "being imported, which is what the decorator is for" 284 ) 285 chosen = name or queue 286 if chosen in self._registrations: 287 raise AceMQError( 288 f"acemq-fastapi: there is already a consumer called {chosen!r}. Two " 289 "consumers on one queue is a reasonable thing to want; give one of them " 290 "a name=... so they can be told apart" 291 ) 292 registration = ConsumerRegistration( 293 name=chosen, 294 queue=queue, 295 handler=handler, 296 codec=codec, 297 retry=retry, 298 prefetch=prefetch, 299 concurrency=concurrency, 300 declare=declare, 301 tag=tag, 302 args=dict(args or {}), 303 ) 304 self._registrations[chosen] = registration 305 return registration
Registers a handler without the decorator.
What a factory function or a plugin uses. Same arguments as
consumer(), and the same restriction: before the lifespan opens.
async def
start_consumer(self, name: str) -> acemq_amqp.connection.Consumer:
307 async def start_consumer(self, name: str) -> Consumer: 308 """Subscribes one registered consumer that is not running. 309 310 What :attr:`consumers` is for in the other direction: a consumer stopped 311 for back pressure, or one left registered by 312 ``acemq.consumer.auto_start=false``, is started again by name. 313 """ 314 registration = self._registrations.get(name) 315 if registration is None: 316 known = ", ".join(sorted(self._registrations)) or "none" 317 raise AceMQError( 318 f"acemq-fastapi: no consumer is registered as {name!r}; registered: {known}" 319 ) 320 if name in self._consumers: 321 return self._consumers[name] 322 consumer = await self._subscribe(registration) 323 self._consumers[name] = consumer 324 return consumer
Subscribes one registered consumer that is not running.
What consumers is for in the other direction: a consumer stopped
for back pressure, or one left registered by
acemq.consumer.auto_start=false, is started again by name.
async def
stop_consumer(self, name: str) -> None:
326 async def stop_consumer(self, name: str) -> None: 327 """Drains and closes one consumer, leaving it registered. 328 329 It finishes what its handlers are holding, exactly as a shutdown does. 330 """ 331 consumer = self._consumers.pop(name, None) 332 if consumer is not None: 333 await consumer.close()
Drains and closes one consumer, leaving it registered.
It finishes what its handlers are holding, exactly as a shutdown does.
def
publisher( self, exchange: str = '', routing_key: str = '', *, codec: acemq_amqp.codec.Codec | None = None, persistent: bool = True, mandatory: bool = False) -> acemq_amqp.connection.Publisher:
337 def publisher( 338 self, 339 exchange: str = "", 340 routing_key: str = "", 341 *, 342 codec: Codec | None = None, 343 persistent: bool = True, 344 mandatory: bool = False, 345 ) -> Publisher: 346 """A publisher for a destination, built once and reused. 347 348 Publishers are cheap to keep and safe to share, and building one per 349 request is a small waste repeated a lot. This caches by every argument 350 that changes what a publisher does, so two routes asking for the same 351 destination get the same object and a route asking for a different one 352 does not quietly get somebody else's. 353 354 :raises AceMQError: before the lifespan has opened the connection 355 """ 356 key = (exchange, routing_key, id(codec) if codec else None, persistent, mandatory) 357 publisher = self._publishers.get(key) 358 if publisher is None: 359 publisher = self.connection.publisher( 360 exchange, 361 routing_key, 362 codec=codec, 363 persistent=persistent, 364 mandatory=mandatory, 365 ) 366 self._publishers[key] = publisher 367 return publisher
A publisher for a destination, built once and reused.
Publishers are cheap to keep and safe to share, and building one per request is a small waste repeated a lot. This caches by every argument that changes what a publisher does, so two routes asking for the same destination get the same object and a route asking for a different one does not quietly get somebody else's.
Raises
- AceMQError: before the lifespan has opened the connection
371 def health_check(self) -> AceMQHealth: 372 """This connection as a :class:`~acemq_amqp.HealthCheck`. 373 374 Pass it to :func:`acemq_amqp.aggregate_health` beside an application's 375 own checks, or mount :meth:`health_router` and let it answer on its own. 376 """ 377 from .health import AceMQHealth 378 379 return AceMQHealth(self)
This connection as a ~acemq_amqp.HealthCheck.
Pass it to acemq_amqp.aggregate_health() beside an application's
own checks, or mount health_router() and let it answer on its own.
async def
health(self) -> acemq_amqp.telemetry.HealthReport:
381 async def health(self) -> acemq_amqp.HealthReport: 382 """What a readiness probe should see. See :class:`~acemq_fastapi.AceMQHealth`.""" 383 return await self.health_check().check()
What a readiness probe should see. See ~acemq_fastapi.AceMQHealth.
def
health_router( self, *, path: str | None = None, tags: list[typing.Any] | None = None) -> Any:
385 def health_router(self, *, path: str | None = None, tags: list[Any] | None = None) -> Any: 386 """An ``APIRouter`` with one route reporting on the connection. 387 388 ``app.include_router(acemq.health_router())`` and nothing else. Mounting 389 it is explicit rather than automatic because a route that appears in an 390 application's OpenAPI document without anybody adding it is a route 391 somebody has to go looking for. 392 """ 393 from .health import health_router 394 395 return health_router(self, path=path, tags=tags)
An APIRouter with one route reporting on the connection.
app.include_router(acemq.health_router()) and nothing else. Mounting
it is explicit rather than automatic because a route that appears in an
application's OpenAPI document without anybody adding it is a route
somebody has to go looking for.
@asynccontextmanager
async def
lifespan( self, app: fastapi.applications.FastAPI | None = None) -> AsyncIterator[Mapping[str, typing.Any]]:
399 @asynccontextmanager 400 async def lifespan(self, app: FastAPI | None = None) -> AsyncIterator[Mapping[str, Any]]: 401 """Opens the connection and the consumers, and drains them on the way out. 402 403 ``FastAPI(lifespan=acemq.lifespan)``. To keep an application's own 404 lifespan as well, see :func:`acemq_fastapi.compose_lifespans` — this one 405 belongs innermost, so the consumers start after whatever they depend on 406 and drain before it goes away. 407 408 The ``app`` is optional so the same context manager can be used in a test 409 or a worker with no ASGI application at all. When one is given it gets 410 ``app.state.acemq``, which is how the request dependencies find this 411 object without a module-level global. 412 """ 413 started = await self.start(app) 414 try: 415 yield {"acemq": self} 416 finally: 417 report = await self.drain() 418 if not report.finished: 419 log.warning( 420 "acemq-fastapi: the drain did not finish in %s; %d message(s) were " 421 "left unsettled and the broker will redeliver them", 422 self._settings.consumer.shutdown_timeout, 423 report.abandoned, 424 ) 425 await self.aclose() 426 log.info( 427 "acemq-fastapi: closed after %.3fs, %d consumer(s) drained", 428 report.waited, 429 started, 430 )
Opens the connection and the consumers, and drains them on the way out.
FastAPI(lifespan=acemq.lifespan). To keep an application's own
lifespan as well, see acemq_fastapi.compose_lifespans() — this one
belongs innermost, so the consumers start after whatever they depend on
and drain before it goes away.
The app is optional so the same context manager can be used in a test
or a worker with no ASGI application at all. When one is given it gets
app.state.acemq, which is how the request dependencies find this
object without a module-level global.
async def
start(self, app: fastapi.applications.FastAPI | None = None) -> int:
432 async def start(self, app: FastAPI | None = None) -> int: 433 """Connects, applies the topology and subscribes the consumers. 434 435 Called by :meth:`lifespan`. Public because a worker process that serves no 436 HTTP at all still wants the same start-up, and should not have to 437 construct an ASGI application to get it. 438 439 :returns: how many consumers were started 440 """ 441 async with self._lock: 442 if self._connection is not None: 443 raise AceMQError("acemq-fastapi: this integration is already started") 444 timeout = self._settings.connection_timeout.total_seconds() 445 try: 446 connection = await asyncio.wait_for(self._connect(self._settings), timeout) 447 except asyncio.TimeoutError as expired: 448 raise AceMQError( 449 f"acemq-fastapi: the broker at {self._settings.broker_url} did not " 450 f"answer within {timeout:g}s" 451 ) from expired 452 self._connection = connection 453 454 try: 455 for publishing in self._on_publish: 456 connection.intercept_publish(publishing) 457 for consuming in self._on_consume: 458 connection.intercept_consume(consuming) 459 460 topology = self._settings.build_topology() 461 if topology is not None: 462 await connection.declare(topology) 463 464 if self._settings.consumer.auto_start: 465 for registration in self._registrations.values(): 466 self._consumers[registration.name] = await self._subscribe(registration) 467 except BaseException: 468 # A half-started application is worse than one that refused to start: 469 # the routes would be up, the consumers would not, and nothing outside 470 # could tell. Undo what was done and let the failure out. 471 await self.drain() 472 await self.aclose() 473 raise 474 475 if app is not None: 476 app.state.acemq = self 477 log.info( 478 "acemq-fastapi: connected to %s, %d consumer(s) running", 479 self._settings.broker_url, 480 len(self._consumers), 481 ) 482 return len(self._consumers)
Connects, applies the topology and subscribes the consumers.
Called by lifespan(). Public because a worker process that serves no
HTTP at all still wants the same start-up, and should not have to
construct an ASGI application to get it.
:returns: how many consumers were started
484 async def drain(self) -> DrainReport: 485 """Stops every consumer, finishing the handlers already running. 486 487 What this finishes, and what it gives back, is written down in the README 488 under *What a drain finishes*. In short: a handler in flight runs to 489 completion; a delivery fetched but not yet started is handed back to the 490 broker; a retry waiting out a short backoff in this process is waited out 491 in full; a publish waiting for its confirm is not waited for. 492 493 Bounded by ``acemq.consumer.shutdown_timeout``. When that expires the 494 handlers still running are cancelled and their messages left unsettled — 495 the broker redelivers them, so the work may be done twice and nothing is 496 dropped. 497 """ 498 if self._connection is None or not self._consumers: 499 self._consumers.clear() 500 return DrainReport(finished=True, waited=0.0, consumers=0) 501 502 consumers = tuple(self._consumers.values()) 503 self._consumers.clear() 504 deadline = self._settings.consumer.shutdown_timeout.total_seconds() 505 loop = asyncio.get_running_loop() 506 started = loop.time() 507 508 closing = [asyncio.create_task(consumer.close()) for consumer in consumers] 509 done, pending = await asyncio.wait(closing, timeout=deadline) 510 waited = loop.time() - started 511 512 if not pending: 513 for task in done: 514 if task.cancelled(): 515 # Its workers were already gone — cancelled from outside, or 516 # killed by something the library logged. Nothing here can 517 # finish a handler that is not running any more. 518 log.warning( 519 "acemq-fastapi: a consumer's workers had already stopped; " 520 "whatever they were holding is back with the broker" 521 ) 522 elif task.exception() is not None: 523 log.warning( 524 "acemq-fastapi: a consumer failed to close: %s", task.exception() 525 ) 526 return DrainReport(finished=True, waited=waited, consumers=len(consumers)) 527 528 # Out of time. Cancelling is the only lever there is, and it is pulled 529 # last rather than first: it aborts the handlers still running and leaves 530 # their deliveries unsettled for the broker to hand on. That is the price 531 # of a deadline, and it is cheaper than being SIGKILLed with the same 532 # messages unsettled and the socket gone. 533 abandoned = sum(consumer.in_flight for consumer in consumers) 534 for task in pending: 535 task.cancel() 536 await asyncio.gather(*pending, return_exceptions=True) 537 return DrainReport( 538 finished=False, 539 waited=loop.time() - started, 540 consumers=len(consumers), 541 abandoned=abandoned, 542 )
Stops every consumer, finishing the handlers already running.
What this finishes, and what it gives back, is written down in the README under What a drain finishes. In short: a handler in flight runs to completion; a delivery fetched but not yet started is handed back to the broker; a retry waiting out a short backoff in this process is waited out in full; a publish waiting for its confirm is not waited for.
Bounded by acemq.consumer.shutdown_timeout. When that expires the
handlers still running are cancelled and their messages left unsettled —
the broker redelivers them, so the work may be done twice and nothing is
dropped.
async def
aclose(self) -> None:
544 async def aclose(self) -> None: 545 """Releases the connection. Idempotent, and safe after a failed start.""" 546 connection = self._connection 547 self._connection = None 548 self._publishers.clear() 549 self._consumers.clear() 550 if connection is not None: 551 with suppress(Exception): 552 await connection.close()
Releases the connection. Idempotent, and safe after a failed start.
AceMQConnection =
typing.Annotated[acemq_amqp.connection.Connection, Depends(dependency=<function get_connection>, use_cache=True, scope=None)]
AceMQDep =
typing.Annotated[AceMQ, Depends(dependency=<function get_acemq>, use_cache=True, scope=None)]
class
AceMQHealth:
40class AceMQHealth: 41 """The connection, as something a probe can ask. 42 43 Up means the broker answered a round trip and every consumer is still reading. 44 Degraded means it answered but a consumer has stopped reading — worth an alert, 45 not worth a restart, because a replacement would almost certainly stall the same 46 way. Down means the broker did not answer at all. 47 48 **A blocked connection is up, with the reason.** That is the one place this 49 disagrees with a naive reading of the broker's state, and it is deliberate: a 50 blocked connection is the broker protecting itself, and an application that 51 fails its own readiness check for it is an application an orchestrator restarts 52 into the same blocked broker, having thrown away whatever it was holding. It is 53 the same call the Spring Boot starter makes, for the same reason. 54 55 The wording of that detail comes from the library and is the same sentence in 56 every AceMQ language, so one alert rule matches a blocked broker whatever the 57 service is written in. This adds ``parts`` and passes the rest through; it does 58 not paraphrase. 59 60 ``acemq.health.timeout`` is handed to the library's probe rather than wrapped 61 around it. The probe has had a deadline of its own since acemq-amqp 0.7.0, and 62 a second one outside it would be the cruder of the two: a blocked broker stops 63 reading its socket, and the library answers that case by re-reading the blocked 64 state and reporting *up* — an outer timeout firing first would turn that answer 65 into a spurious *down*, and would cancel a request the library deliberately 66 abandons instead, because a broker that is not reading cannot be relied on to 67 process a cancellation either. 68 """ 69 70 def __init__(self, acemq: AceMQ, *, name: str = "acemq") -> None: 71 self._acemq = acemq 72 self._name = name 73 74 @property 75 def name(self) -> str: 76 """What this check is called in a combined report.""" 77 return self._name 78 79 async def check(self) -> HealthReport: 80 """The current state. Returns quickly and does not raise.""" 81 acemq = self._acemq 82 if not acemq.started: 83 return HealthReport( 84 HealthStatus.DOWN, 85 "the connection has not been opened; the lifespan is not running", 86 ) 87 88 connection = acemq.connection 89 timeout = acemq.settings.health.timeout.total_seconds() 90 try: 91 report = await connection.health(timeout) 92 except Exception as failure: 93 # A check must not raise; a probe wants an answer, not a 500. 94 return HealthReport( 95 HealthStatus.DOWN, 96 f"the check itself failed: {failure}", 97 parts={"blocked": connection.blocked, **self._parts()}, 98 ) 99 100 return HealthReport( 101 report.status, 102 report.detail, 103 report.checked, 104 {**dict(report.parts), **self._parts()}, 105 ) 106 107 def _parts(self) -> dict[str, Any]: 108 """What this package knows that the connection does not. 109 110 The connection counts its consumers; this names them, because a report 111 saying which consumer stalled is the one worth waking up to. ``blocked`` 112 is not here: it comes from the connection, which is the only thing that 113 can answer it. 114 """ 115 acemq = self._acemq 116 consumers = acemq.consumers 117 return { 118 "consumers": {name: consumer.in_flight for name, consumer in consumers.items()}, 119 "registered": [registration.name for registration in acemq.registrations], 120 }
The connection, as something a probe can ask.
Up means the broker answered a round trip and every consumer is still reading. Degraded means it answered but a consumer has stopped reading — worth an alert, not worth a restart, because a replacement would almost certainly stall the same way. Down means the broker did not answer at all.
A blocked connection is up, with the reason. That is the one place this disagrees with a naive reading of the broker's state, and it is deliberate: a blocked connection is the broker protecting itself, and an application that fails its own readiness check for it is an application an orchestrator restarts into the same blocked broker, having thrown away whatever it was holding. It is the same call the Spring Boot starter makes, for the same reason.
The wording of that detail comes from the library and is the same sentence in
every AceMQ language, so one alert rule matches a blocked broker whatever the
service is written in. This adds parts and passes the rest through; it does
not paraphrase.
acemq.health.timeout is handed to the library's probe rather than wrapped
around it. The probe has had a deadline of its own since acemq-amqp 0.7.0, and
a second one outside it would be the cruder of the two: a blocked broker stops
reading its socket, and the library answers that case by re-reading the blocked
state and reporting up — an outer timeout firing first would turn that answer
into a spurious down, and would cancel a request the library deliberately
abandons instead, because a broker that is not reading cannot be relied on to
process a cancellation either.
AceMQHealth(acemq: AceMQ, *, name: str = 'acemq')
name: str
74 @property 75 def name(self) -> str: 76 """What this check is called in a combined report.""" 77 return self._name
What this check is called in a combined report.
async def
check(self) -> acemq_amqp.telemetry.HealthReport:
79 async def check(self) -> HealthReport: 80 """The current state. Returns quickly and does not raise.""" 81 acemq = self._acemq 82 if not acemq.started: 83 return HealthReport( 84 HealthStatus.DOWN, 85 "the connection has not been opened; the lifespan is not running", 86 ) 87 88 connection = acemq.connection 89 timeout = acemq.settings.health.timeout.total_seconds() 90 try: 91 report = await connection.health(timeout) 92 except Exception as failure: 93 # A check must not raise; a probe wants an answer, not a 500. 94 return HealthReport( 95 HealthStatus.DOWN, 96 f"the check itself failed: {failure}", 97 parts={"blocked": connection.blocked, **self._parts()}, 98 ) 99 100 return HealthReport( 101 report.status, 102 report.detail, 103 report.checked, 104 {**dict(report.parts), **self._parts()}, 105 )
The current state. Returns quickly and does not raise.
class
AceMQSettings(pydantic_settings.main.BaseSettings):
333class AceMQSettings(BaseSettings): 334 """Everything ``acemq.*`` can say. 335 336 Built from the environment by default:: 337 338 settings = AceMQSettings() # ACEMQ_URL, ACEMQ_TLS__..., ... 339 settings = AceMQSettings(url="amqp://host/") # or written down 340 341 A FastAPI application that already keeps its own ``BaseSettings`` can nest 342 this one inside it and pass ``settings.acemq`` to :class:`~acemq_fastapi.AceMQ`. 343 """ 344 345 model_config = SettingsConfigDict( 346 env_prefix="ACEMQ_", 347 env_nested_delimiter="__", 348 env_file=".env", 349 env_file_encoding="utf-8", 350 extra="ignore", 351 ) 352 353 #: Where the broker is. ``amqps://`` is encrypted, ``amqp://`` is not. 354 url: str = "amqp://localhost:5672/" 355 356 #: The login, when it is kept out of the URL — which is how a password stays 357 #: out of a connection string that ends up in a log. 358 username: str | None = None 359 password: SecretStr | None = None 360 361 #: The vhost, when it is not in the URL. Given here it replaces the URL's. 362 virtual_host: str | None = None 363 364 #: What to stamp on published messages. Defaults to the library's 365 #: ``acemq@{hostname}``, which names the machine but does not guess the service. 366 client_name: str | None = None 367 368 #: What publishers and consumers encode and decode with unless they say 369 #: otherwise. One of :func:`acemq_amqp.codec_names`. 370 format: str = "json" 371 372 #: How many unacknowledged messages a consumer holds, unless 373 #: ``consumer.prefetch`` or the registration says otherwise. 374 prefetch: int = Field(default=100, ge=1) 375 376 #: Whether a publish waits for the broker's confirmation. The library always 377 #: publishes with confirms, so turning this off is refused rather than 378 #: silently ignored — see the validator below. 379 publisher_confirms: bool = True 380 381 #: How many publishes may be waiting for the broker at once. The eleventh 382 #: thousand-and-first publish waits for room rather than queueing in memory. 383 max_outstanding_publishes: int = Field(default=1000, ge=1) 384 385 #: How long a publish waits for that room before raising. 386 confirm_timeout: Duration = timedelta(seconds=30) 387 388 #: How long :func:`acemq_amqp.connect` is given before the lifespan gives up. 389 connection_timeout: Duration = timedelta(seconds=10) 390 391 tls: TlsSettings = Field(default_factory=TlsSettings) 392 topology: TopologySettings = Field(default_factory=TopologySettings) 393 consumer: ConsumerSettings = Field(default_factory=ConsumerSettings) 394 health: HealthSettings = Field(default_factory=HealthSettings) 395 396 @field_validator("publisher_confirms") 397 @classmethod 398 def _confirms_are_not_optional(cls, value: bool) -> bool: 399 if not value: 400 raise ValueError( 401 "acemq-fastapi: acemq.publisher_confirms cannot be turned off. The Python " 402 "library publishes with confirms always, and a setting that silently did " 403 "nothing would be worse than one that is not there: a service would think " 404 "it had traded safety for throughput and have neither. Take the setting off." 405 ) 406 return value 407 408 @property 409 def broker_url(self) -> str: 410 """The URL to dial, with ``virtual_host`` applied. 411 412 Credentials are *not* applied here: they travel in 413 :meth:`build_security`, which is where the library puts them, so a URL 414 that ends up in a log or an exception carries no password. 415 """ 416 if self.virtual_host is None: 417 return self.url 418 base, _, _ = self.url.partition("?") 419 query = self.url[len(base) :] 420 host = base.rstrip("/") 421 # Split off the path the URL already carried: everything after the 422 # authority is the vhost, and a setting that names one replaces it. 423 scheme, _, rest = host.partition("://") 424 authority, _, _ = rest.partition("/") 425 vhost = self.virtual_host.lstrip("/") 426 return f"{scheme}://{authority}/{vhost}{query}" 427 428 def build_codec(self) -> Codec: 429 """The codec ``format`` names. 430 431 :raises LookupError: when no codec is registered under that name, listing 432 the ones that are 433 """ 434 return codec_by_name(self.format) 435 436 def build_security(self) -> Security | None: 437 """The library's :class:`~acemq_amqp.Security`, or ``None`` when there is 438 nothing to say beyond the URL.""" 439 credentials = None 440 if self.username is not None or self.password is not None: 441 credentials = Credentials( 442 username=self.username or "", 443 secret=self.password.get_secret_value() if self.password else "", 444 ) 445 if credentials is None and not self.tls.describes_tls: 446 return None 447 password = self.tls.client_key_password 448 return Security( 449 certificate_authority=self.tls.certificate_authority, 450 client_certificate=self.tls.client_certificate, 451 client_key=self.tls.client_key, 452 client_key_password=password.get_secret_value() if password else None, 453 server_name=self.tls.server_name, 454 credentials=credentials, 455 verification=Verification(self.tls.verification), 456 allow_development_certificates=self.tls.allow_development_certificates, 457 ) 458 459 def build_retry(self) -> RetryPolicy: 460 """The consumer default retry policy.""" 461 return self.consumer.retry.build() 462 463 def build_topology(self) -> Topology | None: 464 """The topology to apply at startup, or ``None`` when none was declared.""" 465 if self.topology.empty: 466 return None 467 return self.topology.build(self.build_retry())
Everything acemq.* can say.
Built from the environment by default::
settings = AceMQSettings() # ACEMQ_URL, ACEMQ_TLS__..., ...
settings = AceMQSettings(url="amqp://host/") # or written down
A FastAPI application that already keeps its own BaseSettings can nest
this one inside it and pass settings.acemq to ~acemq_fastapi.AceMQ.
model_config =
{'extra': 'ignore', 'arbitrary_types_allowed': True, 'validate_default': True, 'case_sensitive': False, 'env_prefix': 'ACEMQ_', 'env_prefix_target': 'variable', 'nested_model_default_partial_update': False, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_nested_max_split': None, 'env_parse_none_str': None, 'env_parse_enums': None, 'cli_prog_name': None, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_hide_none_type': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_use_class_docs_for_groups': False, 'cli_show_env_vars': False, 'cli_exit_on_error': True, 'cli_prefix': '', 'cli_flag_prefix_char': '-', 'cli_implicit_flags': False, 'cli_ignore_unknown_args': False, 'cli_kebab_case': False, 'cli_shortcuts': None, 'json_file': None, 'json_file_encoding': None, 'yaml_file': None, 'yaml_file_encoding': None, 'yaml_config_section': None, 'toml_file': None, 'secrets_dir': None, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'enable_decoding': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
confirm_timeout: Annotated[datetime.timedelta, BeforeValidator(func=<function _seconds at 0x7f6985bea160>, json_schema_input_type=PydanticUndefined)]
connection_timeout: Annotated[datetime.timedelta, BeforeValidator(func=<function _seconds at 0x7f6985bea160>, json_schema_input_type=PydanticUndefined)]
tls: TlsSettings
topology: TopologySettings
consumer: ConsumerSettings
health: HealthSettings
broker_url: str
408 @property 409 def broker_url(self) -> str: 410 """The URL to dial, with ``virtual_host`` applied. 411 412 Credentials are *not* applied here: they travel in 413 :meth:`build_security`, which is where the library puts them, so a URL 414 that ends up in a log or an exception carries no password. 415 """ 416 if self.virtual_host is None: 417 return self.url 418 base, _, _ = self.url.partition("?") 419 query = self.url[len(base) :] 420 host = base.rstrip("/") 421 # Split off the path the URL already carried: everything after the 422 # authority is the vhost, and a setting that names one replaces it. 423 scheme, _, rest = host.partition("://") 424 authority, _, _ = rest.partition("/") 425 vhost = self.virtual_host.lstrip("/") 426 return f"{scheme}://{authority}/{vhost}{query}"
The URL to dial, with virtual_host applied.
Credentials are not applied here: they travel in
build_security(), which is where the library puts them, so a URL
that ends up in a log or an exception carries no password.
def
build_codec(self) -> acemq_amqp.codec.Codec:
428 def build_codec(self) -> Codec: 429 """The codec ``format`` names. 430 431 :raises LookupError: when no codec is registered under that name, listing 432 the ones that are 433 """ 434 return codec_by_name(self.format)
The codec format names.
Raises
- LookupError: when no codec is registered under that name, listing the ones that are
def
build_security(self) -> acemq_amqp.security.Security | None:
436 def build_security(self) -> Security | None: 437 """The library's :class:`~acemq_amqp.Security`, or ``None`` when there is 438 nothing to say beyond the URL.""" 439 credentials = None 440 if self.username is not None or self.password is not None: 441 credentials = Credentials( 442 username=self.username or "", 443 secret=self.password.get_secret_value() if self.password else "", 444 ) 445 if credentials is None and not self.tls.describes_tls: 446 return None 447 password = self.tls.client_key_password 448 return Security( 449 certificate_authority=self.tls.certificate_authority, 450 client_certificate=self.tls.client_certificate, 451 client_key=self.tls.client_key, 452 client_key_password=password.get_secret_value() if password else None, 453 server_name=self.tls.server_name, 454 credentials=credentials, 455 verification=Verification(self.tls.verification), 456 allow_development_certificates=self.tls.allow_development_certificates, 457 )
The library's ~acemq_amqp.Security, or None when there is
nothing to say beyond the URL.
def
build_retry(self) -> acemq_amqp.retry.RetryPolicy:
459 def build_retry(self) -> RetryPolicy: 460 """The consumer default retry policy.""" 461 return self.consumer.retry.build()
The consumer default retry policy.
def
build_topology(self) -> acemq_amqp.topology.Topology | None:
463 def build_topology(self) -> Topology | None: 464 """The topology to apply at startup, or ``None`` when none was declared.""" 465 if self.topology.empty: 466 return None 467 return self.topology.build(self.build_retry())
The topology to apply at startup, or None when none was declared.
class
BindingSettings(pydantic.main.BaseModel):
257class BindingSettings(BaseModel): 258 """One binding in ``acemq.topology.bindings``.""" 259 260 queue: str 261 exchange: str 262 routing_key: str = ""
One binding in acemq.topology.bindings.
@dataclass(frozen=True, slots=True)
class
ConsumerRegistration:
78@dataclass(frozen=True, slots=True) 79class ConsumerRegistration: 80 """One ``@acemq.consumer`` declaration, before anything is running. 81 82 ``None`` on the per-consumer settings means "whatever ``acemq.consumer.*`` 83 says", resolved when the lifespan opens rather than when the decorator runs, 84 so settings read from the environment still reach a consumer registered at 85 import time. 86 """ 87 88 #: What this consumer is called to :meth:`AceMQ.consumers`. Defaults to the 89 #: queue name, and has to be unique because that is how it is addressed. 90 name: str 91 queue: str 92 handler: Handler 93 codec: Codec | None = None 94 retry: RetryPolicy | None = None 95 prefetch: int | None = None 96 concurrency: int | None = None 97 declare: bool | None = None 98 #: What to call this consumer to the broker. Empty lets the broker name it. 99 tag: str = "" 100 args: Mapping[str, Any] = field(default_factory=dict)
One @acemq.consumer declaration, before anything is running.
None on the per-consumer settings means "whatever acemq.consumer.*
says", resolved when the lifespan opens rather than when the decorator runs,
so settings read from the environment still reach a consumer registered at
import time.
ConsumerRegistration( name: str, queue: str, handler: Callable[acemq_amqp.connection.Message, acemq_amqp.ack.Ack | Awaitable[acemq_amqp.ack.Ack]], codec: acemq_amqp.codec.Codec | None = None, retry: acemq_amqp.retry.RetryPolicy | None = None, prefetch: int | None = None, concurrency: int | None = None, declare: bool | None = None, tag: str = '', args: Mapping[str, typing.Any] = <factory>)
class
ConsumerSettings(pydantic.main.BaseModel):
190class ConsumerSettings(BaseModel): 191 """The defaults every registered consumer starts with. 192 193 A consumer registered with :meth:`~acemq_fastapi.AceMQ.consumer` overrides 194 any of these per queue; what is set here is what the ones that say nothing get. 195 """ 196 197 #: How many unacknowledged messages a consumer holds. It is also the bound on 198 #: how much work a drain has to give back — see the shutdown section of the 199 #: README. 200 prefetch: int = Field(default=100, ge=1) 201 202 #: How many messages one consumer works on at once. One keeps a queue's 203 #: messages in order. 204 concurrency: int = Field(default=1, ge=1) 205 206 #: Declare the queues a failure needs — ``{queue}.dlq``, ``{queue}.parked``, 207 #: the rungs — before subscribing. Turn it off for a login with no 208 #: ``configure`` permission on the vhost. 209 declare: bool = True 210 211 #: Start the consumers when the lifespan opens. Off leaves them registered and 212 #: stopped, for a process that serves HTTP and should not also consume. 213 auto_start: bool = True 214 215 #: How long the lifespan waits for consumers to drain before giving up on 216 #: them and closing the connection anyway. 217 shutdown_timeout: Duration = timedelta(seconds=20) 218 219 #: The ladder. See :class:`RetrySettings`. 220 retry: RetrySettings = Field(default_factory=RetrySettings)
The defaults every registered consumer starts with.
A consumer registered with ~acemq_fastapi.AceMQ.consumer() overrides
any of these per queue; what is set here is what the ones that say nothing get.
shutdown_timeout: Annotated[datetime.timedelta, BeforeValidator(func=<function _seconds at 0x7f6985bea160>, json_schema_input_type=PydanticUndefined)]
retry: RetrySettings
@dataclass(frozen=True, slots=True)
class
DrainReport:
103@dataclass(frozen=True, slots=True) 104class DrainReport: 105 """What a shutdown got through. 106 107 Returned by :meth:`AceMQ.drain` and logged by the lifespan. ``finished`` 108 false means the deadline expired: the handlers still running were cancelled 109 and their messages left unsettled, so the broker will hand them to whoever 110 replaces this process. Nothing is lost; some work may be done twice. 111 """ 112 113 finished: bool 114 #: Seconds spent draining. 115 waited: float 116 #: How many consumers were asked to stop. 117 consumers: int 118 #: How many messages were still in a handler when the deadline expired. 119 abandoned: int = 0
What a shutdown got through.
Returned by AceMQ.drain() and logged by the lifespan. finished
false means the deadline expired: the handlers still running were cancelled
and their messages left unsettled, so the broker will hand them to whoever
replaces this process. Nothing is lost; some work may be done twice.
class
ExchangeSettings(pydantic.main.BaseModel):
247class ExchangeSettings(BaseModel): 248 """One exchange in ``acemq.topology.exchanges``.""" 249 250 name: str 251 kind: Literal["direct", "topic", "fanout", "headers"] = "topic" 252 durable: bool = True 253 auto_delete: bool = False 254 args: dict[str, Any] = Field(default_factory=dict)
One exchange in acemq.topology.exchanges.
class
HealthSettings(pydantic.main.BaseModel):
313class HealthSettings(BaseModel): 314 """The health route, when one is mounted.""" 315 316 #: Path of the route :meth:`~acemq_fastapi.AceMQ.health_router` builds. 317 path: str = "/health/acemq" 318 319 #: How long the broker gets to answer before the check gives up and reports 320 #: down. Handed to the library's probe as its deadline rather than imposed 321 #: around it, so the report an operator reads is the one the connection 322 #: arrived at — a blocked broker that stops answering is still described as 323 #: blocked rather than as silence. Five seconds, against the library's own 324 #: default of three. 325 timeout: Duration = timedelta(seconds=5) 326 327 #: The status code a report that is not healthy answers with. 503 is what a 328 #: readiness probe expects; a degraded or blocked report still answers 200, 329 #: because degraded is not a reason to be taken out of rotation. 330 unhealthy_status_code: int = 503
The health route, when one is mounted.
@runtime_checkable
class
Lifespan42@runtime_checkable 43class Lifespan(Protocol): 44 """What Starlette accepts: something that takes the app and is an async 45 context manager. It may yield nothing, or a mapping merged into the request 46 state.""" 47 48 def __call__(self, app: Any) -> Any: ... # pragma: no cover - a shape, not code
What Starlette accepts: something that takes the app and is an async context manager. It may yield nothing, or a mapping merged into the request state.
Lifespan(*args, **kwargs)
1771def _no_init_or_replace_init(self, *args, **kwargs): 1772 cls = type(self) 1773 1774 if cls._is_protocol: 1775 raise TypeError('Protocols cannot be instantiated') 1776 1777 # Already using a custom `__init__`. No need to calculate correct 1778 # `__init__` to call. This can lead to RecursionError. See bpo-45121. 1779 if cls.__init__ is not _no_init_or_replace_init: 1780 return 1781 1782 # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`. 1783 # The first instantiation of the subclass will call `_no_init_or_replace_init` which 1784 # searches for a proper new `__init__` in the MRO. The new `__init__` 1785 # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent 1786 # instantiation of the protocol subclass will thus use the new 1787 # `__init__` and no longer call `_no_init_or_replace_init`. 1788 for base in cls.__mro__: 1789 init = base.__dict__.get('__init__', _no_init_or_replace_init) 1790 if init is not _no_init_or_replace_init: 1791 cls.__init__ = init 1792 break 1793 else: 1794 # should not happen 1795 cls.__init__ = object.__init__ 1796 1797 cls.__init__(self, *args, **kwargs)
class
QueueSettings(pydantic.main.BaseModel):
223class QueueSettings(BaseModel): 224 """One queue in ``acemq.topology.queues``.""" 225 226 name: str 227 228 #: Survives a broker restart. 229 durable: bool = True 230 231 #: A durable queue is a quorum queue unless it is asked to be something else, 232 #: because the queue type is an argument the broker compares and two 233 #: languages that disagree about it cannot both consume the queue. 234 quorum: bool | None = None 235 236 auto_delete: bool = False 237 exclusive: bool = False 238 239 #: Declare ``{name}.dlq`` and ``{name}.parked`` alongside it and point the 240 #: broker at them. 241 dead_letter: bool = False 242 243 #: Broker-specific arguments. 244 args: dict[str, Any] = Field(default_factory=dict)
One queue in acemq.topology.queues.
class
RetrySettings(pydantic.main.BaseModel):
144class RetrySettings(BaseModel): 145 """The ladder a consumer follows when a handler asks to be tried again. 146 147 ``enabled`` is off, matching the Spring starter: a retry ladder declares rung 148 queues on the broker, and a service should say it wants them rather than 149 discover them. 150 """ 151 152 #: Whether to build a policy at all. Off means :func:`acemq_amqp.no_retry`. 153 enabled: bool = False 154 155 #: ``exponential`` doubles with 20% jitter; ``fixed`` waits the same every time. 156 kind: Literal["exponential", "fixed"] = "exponential" 157 158 #: Total deliveries, the first one included. 159 max_attempts: int = Field(default=3, ge=1) 160 161 #: The wait before the second attempt. 162 initial_delay: Duration = timedelta(seconds=1) 163 164 #: The ceiling however many attempts have passed. Zero for none. 165 max_delay: Duration = timedelta(minutes=1) 166 167 #: Give up on a message older than this, whatever attempt it is on. 168 give_up_after: Duration | None = None 169 170 #: Waits at or above this live on a rung queue in the broker rather than in 171 #: this process. It is a shutdown setting as much as a reliability one: a 172 #: wait held here is a wait a drain has to sit through. 173 wait_in_broker_from: Duration | None = None 174 175 def build(self) -> RetryPolicy: 176 """The library policy these settings describe.""" 177 if not self.enabled: 178 return no_retry() 179 if self.kind == "fixed": 180 policy = fixed_retry(self.max_attempts, self.initial_delay) 181 else: 182 policy = exponential_retry(self.max_attempts, self.initial_delay, self.max_delay) 183 if self.give_up_after is not None: 184 policy = policy.give_up_after(self.give_up_after) 185 if self.wait_in_broker_from is not None: 186 policy = policy.wait_in_broker_from(self.wait_in_broker_from) 187 return policy
The ladder a consumer follows when a handler asks to be tried again.
enabled is off, matching the Spring starter: a retry ladder declares rung
queues on the broker, and a service should say it wants them rather than
discover them.
initial_delay: Annotated[datetime.timedelta, BeforeValidator(func=<function _seconds at 0x7f6985bea160>, json_schema_input_type=PydanticUndefined)]
max_delay: Annotated[datetime.timedelta, BeforeValidator(func=<function _seconds at 0x7f6985bea160>, json_schema_input_type=PydanticUndefined)]
give_up_after: Optional[Annotated[datetime.timedelta, BeforeValidator(func=<function _seconds at 0x7f6985bea160>, json_schema_input_type=PydanticUndefined)]]
wait_in_broker_from: Optional[Annotated[datetime.timedelta, BeforeValidator(func=<function _seconds at 0x7f6985bea160>, json_schema_input_type=PydanticUndefined)]]
def
build(self) -> acemq_amqp.retry.RetryPolicy:
175 def build(self) -> RetryPolicy: 176 """The library policy these settings describe.""" 177 if not self.enabled: 178 return no_retry() 179 if self.kind == "fixed": 180 policy = fixed_retry(self.max_attempts, self.initial_delay) 181 else: 182 policy = exponential_retry(self.max_attempts, self.initial_delay, self.max_delay) 183 if self.give_up_after is not None: 184 policy = policy.give_up_after(self.give_up_after) 185 if self.wait_in_broker_from is not None: 186 policy = policy.wait_in_broker_from(self.wait_in_broker_from) 187 return policy
The library policy these settings describe.
class
TlsSettings(pydantic.main.BaseModel):
95class TlsSettings(BaseModel): 96 """How to verify the broker and what to present to it. 97 98 Maps onto :class:`acemq_amqp.Security`. It is only consulted for an 99 ``amqps://`` URL — the library refuses TLS settings against a plaintext URL 100 rather than ignoring them, and this module passes that refusal straight 101 through. 102 """ 103 104 #: A PEM file holding the authority to verify the broker against. Naming one 105 #: replaces the machine's trust store rather than adding to it. 106 certificate_authority: Path | None = None 107 108 #: A PEM certificate to present, for a broker that authenticates by certificate. 109 client_certificate: Path | None = None 110 111 #: The PEM private key for that certificate. Defaults to the certificate file. 112 client_key: Path | None = None 113 114 #: The passphrase on that key, when it has one. 115 client_key_password: SecretStr | None = None 116 117 #: The name to check the certificate against, when it is not the host in the URL. 118 server_name: str | None = None 119 120 #: ``certificate`` checks the chain and the name. ``nothing-at-all`` checks 121 #: nothing, and is spelled that way so a configuration file cannot hide what 122 #: it is giving up behind a ``verify: false``. 123 verification: Literal["certificate", "nothing-at-all"] = "certificate" 124 125 #: Whether to accept a broker certificate carrying the development marker. 126 #: Off, so a laptop's settings copied into a deployment fails loudly. 127 allow_development_certificates: bool = False 128 129 @property 130 def describes_tls(self) -> bool: 131 """Whether anything here is worth passing to the library.""" 132 return any( 133 ( 134 self.certificate_authority, 135 self.client_certificate, 136 self.client_key, 137 self.server_name, 138 self.verification != "certificate", 139 self.allow_development_certificates, 140 ) 141 )
How to verify the broker and what to present to it.
Maps onto acemq_amqp.Security. It is only consulted for an
amqps:// URL — the library refuses TLS settings against a plaintext URL
rather than ignoring them, and this module passes that refusal straight
through.
describes_tls: bool
129 @property 130 def describes_tls(self) -> bool: 131 """Whether anything here is worth passing to the library.""" 132 return any( 133 ( 134 self.certificate_authority, 135 self.client_certificate, 136 self.client_key, 137 self.server_name, 138 self.verification != "certificate", 139 self.allow_development_certificates, 140 ) 141 )
Whether anything here is worth passing to the library.
class
TopologySettings(pydantic.main.BaseModel):
265class TopologySettings(BaseModel): 266 """What to declare when the lifespan opens. 267 268 Declaring nothing is the default and is a real answer: a service that only 269 consumes queues somebody else owns should not create them. 270 """ 271 272 queues: list[QueueSettings] = Field(default_factory=list) 273 exchanges: list[ExchangeSettings] = Field(default_factory=list) 274 bindings: list[BindingSettings] = Field(default_factory=list) 275 276 #: Whether the rung queues of the consumer retry policy are declared with each 277 #: queue. On, because a policy whose rungs are missing falls back to waiting in 278 #: this process, which is the slow drain nobody expects. 279 declare_retry_rungs: bool = True 280 281 @property 282 def empty(self) -> bool: 283 """Whether there is anything at all to apply.""" 284 return not (self.queues or self.exchanges or self.bindings) 285 286 def build(self, retry: RetryPolicy | None = None) -> Topology: 287 """The library topology these settings describe.""" 288 topology = Topology() 289 for exchange in self.exchanges: 290 topology.exchange( 291 exchange.name, 292 exchange.kind, 293 durable=exchange.durable, 294 auto_delete=exchange.auto_delete, 295 args=exchange.args, 296 ) 297 for queue in self.queues: 298 topology.queue( 299 queue.name, 300 durable=queue.durable, 301 auto_delete=queue.auto_delete, 302 exclusive=queue.exclusive, 303 quorum=queue.quorum, 304 dead_letter=queue.dead_letter, 305 retry=retry if self.declare_retry_rungs else None, 306 args=queue.args, 307 ) 308 for binding in self.bindings: 309 topology.binding(binding.queue, binding.exchange, binding.routing_key) 310 return topology
What to declare when the lifespan opens.
Declaring nothing is the default and is a real answer: a service that only consumes queues somebody else owns should not create them.
queues: list[QueueSettings]
exchanges: list[ExchangeSettings]
bindings: list[BindingSettings]
empty: bool
281 @property 282 def empty(self) -> bool: 283 """Whether there is anything at all to apply.""" 284 return not (self.queues or self.exchanges or self.bindings)
Whether there is anything at all to apply.
def
build( self, retry: acemq_amqp.retry.RetryPolicy | None = None) -> acemq_amqp.topology.Topology:
286 def build(self, retry: RetryPolicy | None = None) -> Topology: 287 """The library topology these settings describe.""" 288 topology = Topology() 289 for exchange in self.exchanges: 290 topology.exchange( 291 exchange.name, 292 exchange.kind, 293 durable=exchange.durable, 294 auto_delete=exchange.auto_delete, 295 args=exchange.args, 296 ) 297 for queue in self.queues: 298 topology.queue( 299 queue.name, 300 durable=queue.durable, 301 auto_delete=queue.auto_delete, 302 exclusive=queue.exclusive, 303 quorum=queue.quorum, 304 dead_letter=queue.dead_letter, 305 retry=retry if self.declare_retry_rungs else None, 306 args=queue.args, 307 ) 308 for binding in self.bindings: 309 topology.binding(binding.queue, binding.exchange, binding.routing_key) 310 return topology
The library topology these settings describe.
__version__ =
'0.1.5'
51def compose_lifespans(*lifespans: Lifespan) -> Lifespan: 52 """Runs several lifespans as one, outermost first. 53 54 The state each one yields is merged, in order, so a later lifespan can 55 override an earlier one's key and a route sees all of them. A lifespan that 56 yields nothing contributes nothing, which is the common case. 57 58 An exception on the way up unwinds the ones already open, because 59 :class:`~contextlib.AsyncExitStack` is doing the work and that is what it is 60 for — half an application started is worse than none. 61 62 :param lifespans: the lifespans, outside in 63 :returns: one lifespan to pass to ``FastAPI(lifespan=...)`` 64 """ 65 66 @asynccontextmanager 67 async def composed(app: Any) -> AsyncIterator[Mapping[str, Any]]: 68 state: dict[str, Any] = {} 69 async with AsyncExitStack() as stack: 70 for lifespan in lifespans: 71 yielded = await stack.enter_async_context(lifespan(app)) 72 if isinstance(yielded, Mapping): 73 state.update(yielded) 74 yield state 75 76 return composed
Runs several lifespans as one, outermost first.
The state each one yields is merged, in order, so a later lifespan can override an earlier one's key and a route sees all of them. A lifespan that yields nothing contributes nothing, which is the common case.
An exception on the way up unwinds the ones already open, because
~contextlib.AsyncExitStack is doing the work and that is what it is
for — half an application started is worse than none.
Parameters
- lifespans: the lifespans, outside in
:returns: one lifespan to pass to
FastAPI(lifespan=...)
51def get_acemq(request: Request) -> AceMQ: 52 """The integration for the application handling this request. 53 54 :raises AceMQError: when the application was built without 55 ``lifespan=acemq.lifespan``, which is the mistake this is really here to 56 name. The alternative is an ``AttributeError`` on ``app.state`` that says 57 nothing about what was forgotten 58 """ 59 acemq: AceMQ | None = getattr(request.app.state, "acemq", None) 60 if acemq is None: 61 raise AceMQError( 62 "acemq-fastapi: this application has no AceMQ on app.state. Build it with " 63 "FastAPI(lifespan=acemq.lifespan), or compose_lifespans(..., acemq.lifespan) " 64 "when it already has a lifespan of its own" 65 ) 66 return acemq
The integration for the application handling this request.
Raises
- AceMQError: when the application was built without
lifespan=acemq.lifespan, which is the mistake this is really here to name. The alternative is anAttributeErroronapp.statethat says nothing about what was forgotten
def
get_connection( acemq: Annotated[AceMQ, Depends(dependency=<function get_acemq>, use_cache=True, scope=None)]) -> acemq_amqp.connection.Connection:
69def get_connection(acemq: Annotated[AceMQ, Depends(get_acemq)]) -> Connection: 70 """The open connection, for a route that needs more than a publisher. 71 72 Reading a queue depth, declaring something on the fly, pulling one message — 73 the things :class:`~acemq_amqp.Publisher` deliberately does not do. 74 """ 75 return acemq.connection
The open connection, for a route that needs more than a publisher.
Reading a queue depth, declaring something on the fly, pulling one message —
the things ~acemq_amqp.Publisher deliberately does not do.
def
health_router( acemq: AceMQ, *, path: str | None = None, tags: list[str | enum.Enum] | None = None) -> fastapi.routing.APIRouter:
142def health_router( 143 acemq: AceMQ, *, path: str | None = None, tags: list[str | Enum] | None = None 144) -> APIRouter: 145 """A router with one route reporting on the connection. 146 147 ``app.include_router(acemq.health_router())``. The path is 148 ``acemq.health.path``, ``/health/acemq`` unless it is set. 149 150 The status code is the part worth reading twice: 200 for up **and for 151 degraded and blocked**, ``acemq.health.unhealthy_status_code`` (503) only for 152 down. A readiness probe pointed at this route takes the instance out of 153 rotation when the broker is unreachable and leaves it in when the broker is 154 merely unhappy, which is the behaviour described in 155 :class:`AceMQHealth`. An alert should read the body, not the code. 156 """ 157 router = APIRouter(tags=tags if tags is not None else ["acemq"]) 158 159 @router.get( 160 path or acemq.settings.health.path, 161 summary="AceMQ broker connection", 162 response_class=JSONResponse, 163 ) 164 async def acemq_health() -> JSONResponse: 165 report = await acemq.health() 166 code = 200 if report.healthy else acemq.settings.health.unhealthy_status_code 167 return JSONResponse(report_as_json(report), status_code=code) 168 169 return router
A router with one route reporting on the connection.
app.include_router(acemq.health_router()). The path is
acemq.health.path, /health/acemq unless it is set.
The status code is the part worth reading twice: 200 for up and for
degraded and blocked, acemq.health.unhealthy_status_code (503) only for
down. A readiness probe pointed at this route takes the instance out of
rotation when the broker is unreachable and leaves it in when the broker is
merely unhappy, which is the behaviour described in
AceMQHealth. An alert should read the body, not the code.
def
publishes( routing_key: str = '', exchange: str = '', *, codec: acemq_amqp.codec.Codec | None = None, persistent: bool = True, mandatory: bool = False) -> Any:
78def publishes( 79 routing_key: str = "", 80 exchange: str = "", 81 *, 82 codec: Codec | None = None, 83 persistent: bool = True, 84 mandatory: bool = False, 85) -> Any: 86 """Builds a dependency yielding a publisher for one destination. 87 88 The routing key comes first because publishing straight to a queue is the 89 common case: with no exchange, the broker's default exchange routes to the 90 queue whose name matches the key. 91 92 The publisher is built once per destination and reused, so a route that is 93 called a thousand times a second does not build a thousand publishers. It is 94 a dependency rather than a module-level object because it has to be resolved 95 after the lifespan has opened the connection, and because a test that builds 96 a second application in the same process must not be handed the first one's. 97 98 :param routing_key: what to publish under, or a queue name 99 :param exchange: where to publish, empty for the default exchange 100 :param codec: a codec other than the connection's 101 :param persistent: ask the broker to write these messages to disk 102 :param mandatory: fail when a message reaches no queue at all, rather than 103 letting the broker drop it silently 104 :returns: something to put in ``Depends(...)`` 105 """ 106 107 def dependency(acemq: Annotated[AceMQ, Depends(get_acemq)]) -> Publisher: 108 return acemq.publisher( 109 exchange, 110 routing_key, 111 codec=codec, 112 persistent=persistent, 113 mandatory=mandatory, 114 ) 115 116 return dependency
Builds a dependency yielding a publisher for one destination.
The routing key comes first because publishing straight to a queue is the common case: with no exchange, the broker's default exchange routes to the queue whose name matches the key.
The publisher is built once per destination and reused, so a route that is called a thousand times a second does not build a thousand publishers. It is a dependency rather than a module-level object because it has to be resolved after the lifespan has opened the connection, and because a test that builds a second application in the same process must not be handed the first one's.
Parameters
- routing_key: what to publish under, or a queue name
- exchange: where to publish, empty for the default exchange
- codec: a codec other than the connection's
- persistent: ask the broker to write these messages to disk
- mandatory: fail when a message reaches no queue at all, rather than
letting the broker drop it silently
:returns: something to put in
Depends(...)
def
report_as_json(report: acemq_amqp.telemetry.HealthReport) -> dict[str, typing.Any]:
123def report_as_json(report: HealthReport) -> dict[str, Any]: 124 """A health report as the shape the route answers with.""" 125 return { 126 "status": report.status.value, 127 "healthy": report.healthy, 128 "detail": report.detail, 129 "checked": report.checked.isoformat(), 130 "parts": _plain(dict(report.parts)), 131 }
A health report as the shape the route answers with.