API 参考#

binomic.client#

class binomic.client.Binomic(broker_dsn: str, module_name: str, config: BinomicConfig)#

Binomic working client.

Entering the async context starts the master process group and keeps it running until the context exits.

async submit(msg: Message) → UUID#

Dispatch a message to its task.

direct tasks are enqueued immediately and delay tasks are handed to the scheduler for a deferred enqueue. Scheduled tasks (cron and interval) cannot be submitted: they are registered automatically when the client starts.

Raises:: TaskNotFoundError: if no task is registered under the message's name. RuntimeError: if the client has not been started with arun. ValueError: if the task's mode is neither direct nor delay.

class binomic.client.BinomicFactory(broker_dsn: str, module_name: str, config: BinomicConfig)#

Binomic working client factory.

create() → Binomic#: Return the client, creating it on the first call.

binomic.config#

class binomic.config.BinomicConfig(queues: list[str], workers: int = 1, concurrency: int = 5, max_attempts: int = 3)#

Binomic configuration.

workers is the number of worker subprocesses the master spawns and concurrency the number of tasks each worker executes concurrently; every stream in queues is consumed by all workers. max_attempts is how many times a message may be delivered before it is dropped.

binomic.task#

exception binomic.task.DuplicateTaskError#: Duplicate task error.

exception binomic.task.TaskNotFoundError#: Task not found error.

class binomic.task.TaskScheduler(timezone: tzinfo = datetime.timezone.utc)#

Scheduler class for Binomic tasks.

cron(func: Callable[[...], Any | Awaitable[Any]], spec: TaskSpec, *, args: Sequence[Any] | None = None, kwargs: Mapping[str, Any] | None = None) → None#: Schedule a task to run at specific times.

delay(func: Callable[[...], Any | Awaitable[Any]], spec: TaskSpec, *, delay: float | None = None, args: Sequence[Any] | None = None, kwargs: Mapping[str, Any] | None = None) → None#: Delay a task.

interval(func: Callable[[...], Any | Awaitable[Any]], spec: TaskSpec, *, args: Sequence[Any] | None = None, kwargs: Mapping[str, Any] | None = None) → None#: Schedule a task to run at regular intervals.

shutdown() → None#: Shutdown the scheduler.

start() → None#: Start the scheduler.

class binomic.task.TaskSpec(name: str, queue: str, fn: Callable[[P], T], mode: Literal['direct', 'delay', 'cron', 'interval'] = 'direct', delay: float | None = None, cron: str | None = None, interval: float | None = None)#

binomic.task.autodiscover(package: str, *, on_error: Literal['warn', 'raise'] = 'warn') → list[str]#

Import every tasks module inside package to trigger registration.

Returns the names of the imported modules; a failing module import is warned or raised depending on on_error.

binomic.task.task(queue: str, *, mode: Literal['direct']) → Callable[[Callable[P, T]], TaskSpec[P, T]]#

binomic.task.task(queue: str, *, mode: Literal['delay'], delay: float) → Callable[[Callable[P, T]], TaskSpec[P, T]]

binomic.task.task(queue: str, *, mode: Literal['cron'], cron: str) → Callable[[Callable[P, T]], TaskSpec[P, T]]

binomic.task.task(queue: str, *, mode: Literal['interval'], interval: float) → Callable[[Callable[P, T]], TaskSpec[P, T]]

Declare a task and register it under the function's lowercase name.

Mode direct executes on submit, delay schedules execution delay seconds after submit, cron repeats execution on the given cron expression, and interval repeats it every interval seconds.

Raises:

ValueError: if delay, cron or interval mode lacks its: required argument, or if a cron or interval task's function takes parameters.

binomic.task.scheduler#

class binomic.task.scheduler.TaskScheduler(timezone: tzinfo = datetime.timezone.utc)#

Scheduler class for Binomic tasks.

cron(func: Callable[[...], Any | Awaitable[Any]], spec: TaskSpec, *, args: Sequence[Any] | None = None, kwargs: Mapping[str, Any] | None = None) → None#: Schedule a task to run at specific times.

delay(func: Callable[[...], Any | Awaitable[Any]], spec: TaskSpec, *, delay: float | None = None, args: Sequence[Any] | None = None, kwargs: Mapping[str, Any] | None = None) → None#: Delay a task.

interval(func: Callable[[...], Any | Awaitable[Any]], spec: TaskSpec, *, args: Sequence[Any] | None = None, kwargs: Mapping[str, Any] | None = None) → None#: Schedule a task to run at regular intervals.

shutdown() → None#: Shutdown the scheduler.

start() → None#: Start the scheduler.

binomic.message#

class binomic.message.Message(name: str, id: UUID = <factory>, attempt: int = 1, args: list[Any] = <factory>, kwargs: dict[str, ~typing.Any]=<factory>)#

Message for binomic.

name is the registered task name and args / kwargs carry the task call arguments. The delivery timestamp is not carried here: the broker stamps it on the stream entry when the message is enqueued.

classmethod from_json(json_str: str) → Message#: Create a data structure from a JSON string.

to_json() → str#: Convert the data structure to a JSON string.

binomic.broker#

class binomic.broker.AsyncredisBroker(dsn: str, queues: Sequence[str], **config: Any)#

Broker based on async redis.

get_stream_key(queue: str) → str#: Namespace a queue name into its Redis stream key.

class binomic.broker.Broker(*args, **kwargs)#: Broker protocol class.

class binomic.broker.BrokerFactory(dsn: str, queues: list[str])#: A factory class for creating broker instances.

class binomic.broker.Entry(queue: str, msg_id: str, fields: Fields)#

Entry in the broker.

fields: Fields#: Alias for field number 2

msg_id: str#: Alias for field number 1

queue: str#: Alias for field number 0

binomic.worker#

class binomic.worker.Master(*, broker_dsn: str, module_name: str, policy: MasterPolicy)#: Binomic processing visor.

class binomic.worker.MasterPolicy(workers: int, worker: WorkerPolicy)#: Binomic master policy.

class binomic.worker.MasterPresence#

Master-side view of the liveness of every worker.

One instance covers the whole worker set, the way the map it replaced did: presence() returns every watched worker's last beat. Comparing that map against the clock is how the visor tells a worker that is stuck apart from one that is merely idle: is_alive stays true while a stalled worker stops writing.

presence() → dict[str, float]#

The last beat of every watched worker.

Each cell is read through its object, without taking its lock: the lock exists for read-modify-write sequences and nothing here does one. Taking it would mean waiting on a worker that died -- or was stopped -- while holding it, which would wedge the visor for good.

watch(ident: str, cell: Synchronized[float]) → None#: Take over the cell a worker beats on, replacing any earlier one.

class binomic.worker.SubprocessPresence(heartbeat: Synchronized[float])#

Subprocess-side view of a worker's liveness.

Borrows the cell the master created: the master owns it and frees it, so nothing here closes it.

class binomic.worker.Worker(*, broker_dsn: str, module_name: str, consumer: str, policy: WorkerPolicy, presence: SubprocessPresence)#

Binomic subprocessing worker.

Spawned rather than forked: a forked child inherits the parent's running event loop, so anyio.run cannot start in it.

run() → None#: Method to be run in sub-process; can be overridden in sub-class

class binomic.worker.WorkerPolicy(queues: list[str], concurrency: int, task_timeout: float = 600.0, max_attempts: int = 3, read_count: int = 10, poll_interval: float = 0.1, heartbeat_interval: float = 5.0, reclaim_interval: float = 30.0)#: Binomic worker policy.

binomic.base#

class binomic.base.BaseStruct#

Base data structure.

classmethod from_json(json_str: str) → Self#: Create a data structure from a JSON string.

to_json() → str#: Convert the data structure to a JSON string.

exception binomic.base.BinomicError#: Base class for binomic exceptions.

exception binomic.base.DeserializationError#: Exception raised for errors during deserialization.

exception binomic.base.SerializationError#: Exception raised for errors during serialization.

binomic.plugin.litestar#

class binomic.plugin.litestar.BinomicPlugin(app_name: str, broker_dsn: str, config: BinomicConfig)#

Binomic plugin for litestar.

on_app_init(app_config: AppConfig) → AppConfig#

Receive the AppConfig instance after on_app_init hooks have been called.

Examples:

from litestar import Litestar, get
from litestar.di import Provide
from litestar.plugins import InitPluginProtocol


def get_name() -> str:
    return "world"


@get("/my-path")
def my_route_handler(name: str) -> dict[str, str]:
    return {"hello": name}


class MyPlugin(InitPluginProtocol):
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        app_config.dependencies["name"] = Provide(get_name)
        app_config.route_handlers.append(my_route_handler)
        return app_config


app = Litestar(plugins=[MyPlugin()])

Args:

app_config: The AppConfig instance.

Returns:

The app config object.