amgi-common
This package includes some useful helpers for writing AMGI servers.
Installation
pip install amgi-common==0.48.0
Constructs
Lifespan
This class handles the AMGI lifespan protocol.
from amgi_common import Lifespan
async def main_loop(app, state):
"""Handle event batches"""
async def serve(app):
async with Lifespan(app) as state:
# handle app calls
await main_loop(app, state)
Stoppable
This class should help with graceful shutdowns. Whenever you have a call to something with a timeout, it cancels that task. .
For example, if you have a client that has a method fetch_messages, which has a timeout you could loop like so:
from amgi_common import Stoppable
class Server:
def __init__(self, app):
self._app = app
self._stoppable = Stoppable()
self._client = Client()
self._running = True
async def main_loop(self, app, state):
while self._running:
messages = await self._client.fetch_messages(timeout=10_000)
# Handle messages
def stop(self):
self._running = False
There are several ways to deal with this:
- The above example, where on stop you will have to wait for the timeout plus the time to handle messages
- Run the main loop in a task, and cancel it on stop
Both have their own problems. In the first case, you could be waiting a long time. In the second case, when you receive messages, you should probably process them before shutting down.
The class Stoppable is there to help you. It will return the results of the callable iterably, but if the stoppable has
been told to stop, it will cancel any tasks it has running in the background, and stop any current iterations.
from amgi_common import Stoppable
class Server:
def __init__(self, app):
self._app = app
self._stoppable = Stoppable()
self._client = Client()
async def main_loop(self, app, state):
async for messages in self._stoppable.call(
self._client.fetch_messages, timeout=10_000
):
# Handle messages
pass
def stop(self):
self._stoppable.stop()
OperationBatcher
This class can simplify batch operations, for example, publishing multiple messages simultaneous. The implementation allows you to enqueue an item to run the operation on, and the operation will be scheduled in batches:
from amgi_common import OperationBatcher
async def batch_operation(items):
# Do a batch operation to get the batch result
return [item for item in batch_result]
operation_batcher = OperationBatcher(batch_operation)
# You can enqueue an item, it will be processed in the background automatically
result = await operation_batcher.enqueue({"an": "item"})
The batch operation should return an iterable of results, or exceptions. This allows for per item errors. The batch sizes can also be limited, which is usefully when APIs have a hard limit.
Contact
For questions or suggestions, please contact jack.burridge@mail.com.
License
Copyright 2025 AMGI
1 maintainer
Metadata
1 maintainer
Release files for amgi-common 0.48.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| amgi_common-0.48.0-py3-none-any.whl | Python 3 | none | any | Details |
Release files / amgi_common-0.48.0-py3-none-any.whl
| Download URL | amgi_common-0.48.0-py3-none-any.whl |
|---|---|
| Size | 6.3 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
1b73ca7bb7bfe8bbf9a269f84f9b96d39585aab1215124b8005102452f935dac
|
|
BLAKE2b-256 checksum How to use checksums |
d7a17fb8e5063ef635f1a8b91aa8887e3f53ee838e546ebd30a1a20f7d1b82ab
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.0 {"installer":{"name":"uv","version":"0.13.0","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
