aiomisc

The aiomisc library provides a collection of miscellaneous utilities for asyncio. The following modules are available for use:

• aiomisc.aggregate: Utilities for aggregating results.
• aiomisc.backoff: Implementation of backoff strategies.
• aiomisc.circuit_breaker: Circuit breaker patterns for fault tolerance.
• aiomisc.compat: Compatibility utilities.
• aiomisc.context: Context management utilities.
• aiomisc.counters: Async-friendly counters.
• aiomisc.cron: Cron-like scheduling.
• aiomisc.entrypoint: Tools for managing application entrypoints.
• aiomisc.io: I/O related utilities.
• aiomisc.iterator_wrapper: Wrappers for iterators.
• aiomisc.log: Logging utilities.
• aiomisc.periodic: Periodic task execution.
• aiomisc.plugins: Plugin system utilities.
• aiomisc.pool: General pooling utilities.
• aiomisc.process_pool: Process pool management.
• aiomisc.recurring: Recurring task management.
• aiomisc.signal: Signal handling utilities.
• aiomisc.thread_pool: Thread pool management.
• aiomisc.timeout: Timeout management utilities.
• aiomisc.utils: General purpose utilities.
• aiomisc.worker_pool: Worker pool management.

The Service abstraction is an abstract base class used to organize asynchronous programs into independent, concurrent services with well-defined start and stop logic. It helps manage the lifecycle of components (like database connections or web servers) and simplifies testing and error handling.

There are two primary modes of operation for a Service:

1. Infinite/Long-running Mode: The start() method runs indefinitely (e.g., a loop). You must signal that initialization is complete by calling self.start_event.set(). Stopping the service is achieved by completing the coroutine created by start().
2. Explicit Start/Stop Mode: You implement both start() and stop() methods. The service is started once and stopped once via the explicit stop() call.

import asyncio
import aiomisc

# Mode 1: Infinite Service
class InfinityService(aiomisc.Service):
    async def start(self):
        self.start_event.set()  # Signal initialization is done
        while True:
            await asyncio.sleep(1)

# Mode 2: Explicit Start/Stop Service
class OrdinaryService(aiomisc.Service):
    async def start(self):
        # do some stuff
        ...

    async def stop(self, exception: Exception = None) -> Any:
        # do some stuff
        ...

The Service class uses a metaclass to handle configuration via class attributes. This allows for easy dependency injection and validation at the class declaration stage.

• Automatic Assignment: Any keyword arguments passed to the service constructor are automatically assigned to self as attributes.
• __required__: A tuple of attribute names that must be provided during service initialization. If they are missing, an error is raised.
• __async_required__: A tuple of method names that must be explicitly defined as asynchronous (async def). This is useful for enforcing interfaces in base classes.

If a subclass violates these requirements (e.g., defines a required async method as a regular function), a TypeError is raised during class declaration.

import aiomisc

class HelloService(aiomisc.Service):
    __required__ = ("name", "title")
    __async_required__ = ("greeting",)

    name: str
    title: str

    async def greeting(self) -> str:
        return f"Hello {self.title} {self.name}"

    async def start(self):
        print(await self.greeting())

The aiomisc.CircuitBreaker implements the circuit breaker design pattern to detect failures and prevent them from constantly recurring during maintenance or external system failures. It tracks statistics of successful and failed calls to determine if it should 'break' the circuit.

To use it, you can either wrap calls manually using the .context() context manager or use the cutout decorator for a more ergonomic approach.

Note: When the circuit is in a BROKEN or RECOVERING state, it raises a CircuitBroken exception. This exception is not counted towards the error statistics.

from aiomisc import CircuitBreaker
import aiohttp

# Configuration example
# If 20% errors occur within 20 seconds, the circuit breaks for 5 seconds
cb = CircuitBreaker(
    error_ratio=0.2,
    response_time=20,
    exceptions=[aiohttp.ClientError],
    broken_time=5
)

# Usage via context manager
async def my_function(session):
    with cb.context():
        async with session.get('https://api.example.com') as response:
            return await response.text()

A Service is a class derived from aiomisc.Service. Services are managed by the entrypoint and are started concurrently.

Lifecycle

1. Start: Implement async def start(self) -> None:.
2. Stop: Optionally implement async def stop(self, exc: Optional[Exception]) -> None:. This is called when the entrypoint context manager exits.

Signaling Completion

If your start method is a long-running task (a payload) and you don't want to implement a stop method, you must notify the entrypoint that initialization is complete by calling self.start_event.set(). Otherwise, the entrypoint will wait indefinitely for the service to 'finish' starting before yielding control to the context manager body.

Implementation Example

from aiomisc import entrypoint, Service
import asyncio

class MyService(Service):
    async def start(self):
        # Perform initialization
        print("Service starting...")
        # Notify entrypoint that we are ready
        self.start_event.set()
        # Keep the service running
        await asyncio.sleep(3600)

    async def stop(self, exc):
        print("Service stopping...")

with entrypoint(MyService()) as loop:
    loop.run_forever()

from aiomisc import entrypoint, Service

class MyService(Service):
    async def start(self):
        do_something_when_start()

    async def stop(self, exc):
        do_graceful_shutdown()


with entrypoint(MyService()) as loop:
    loop.run_forever()

The Service metaclass automatically converts all keyword arguments (kwargs) passed during service initialization into instance attributes.

To enforce that certain parameters are provided during initialization, define a __required__ attribute as a frozenset containing the names of the required keys. You can also define default values for attributes using standard Python type annotations.

import asyncio
from aiomisc import entrypoint
from aiomisc.service import Service

class LoggingService(Service):
    # Define required kwargs
    __required__ = frozenset({'name'})

    # Define default values
    delay: int = 1

    async def start(self):
        self.start_event.set()
        while True:
            # self.name was passed in kwargs
            print('Hello from service', self.name)
            # self.delay uses the default or passed value
            await asyncio.sleep(self.delay)

services = (
    LoggingService(name='#1'),
    LoggingService(name='#2', delay=3),
)

with entrypoint(*services) as loop:
    loop.run_forever()

By default, aiomisc's event_loop and entrypoint fixtures are function-scoped, meaning a fresh loop is created and closed for every test. While safe, this is inefficient for expensive resources like database connection pools or HTTP sessions.

To share resources across multiple tests, you must use a wider fixture scope (e.g., module or session). However, a wider-scoped async fixture must run on an event loop that lives at least as long as the fixture itself.

The Key Rule: The event_loop fixture scope must be greater than or equal to the scope of every async fixture that depends on it.

session >= module >= class >= function

@pytest.fixture(scope="module")
def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        yield loop
    finally:
        loop.close()
        asyncio.set_event_loop(None)

RecurringCallback runs a coroutine function periodically using a user-defined asynchronous strategy function. The strategy function receives the RecurringCallback instance and must return a number (representing the delay in seconds) or raise specific exceptions to control flow.

Control Flow via Exceptions:

• StrategySkip(delay): Skips the current attempt and waits for the specified delay before the next attempt.
• StrategyStop(): Terminates the recurring execution.

Note: If the strategy function returns a non-numeric value or fails to raise these specific exceptions, the execution will be terminated.

from typing import Union
from aiomisc import new_event_loop, RecurringCallback, StrategySkip, StrategyStop

async def callback():
    print("Hello")

async def strategy(_: RecurringCallback) -> Union[int, float]:
    # Example logic for dynamic delays
    # ...
    
    # Skip this attempt and wait 10 seconds
    raise StrategySkip(10)
    
    # Or stop execution entirely
    # raise StrategyStop()

if __name__ == '__main__':
    loop = new_event_loop()
    periodic = RecurringCallback(callback)
    task = periodic.start(strategy)
    loop.run_forever()