amgi-common
This package includes some useful helpers for writing AMGI servers.
Installation
pip install amgi-common==0.47.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
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file amgi_common-0.47.0.tar.gz.
File metadata
- Download URL: amgi_common-0.47.0.tar.gz
- Upload date:
- Size: 5.3 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","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}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
38720dc236dceedfadfd7576ef68e9c177a40d4521cbe371172e4eeb9ee80172
|
|
| MD5 |
ad0b16c7fcac6b23f7b52e841668fcce
|
|
| BLAKE2b-256 |
e6356de1a075f8f6a5856de54341776ae3eb19393929d61dc352c3567ea95241
|
File details
Details for the file amgi_common-0.47.0-py3-none-any.whl.
File metadata
- Download URL: amgi_common-0.47.0-py3-none-any.whl
- Upload date:
- Size: 6.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","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}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
169247abc06bbd8e9287799213e7f135c8c08625dc12b2f4ff4fcd3c8b63f735
|
|
| MD5 |
7673a38929f738cc5e57e0fa92a8c7c2
|
|
| BLAKE2b-256 |
84f9c04e5f22f06fb7a5f0bcbd44d7b4d993c91438590d33bd20551bb7bb8ffe
|
