faust.agents

Agents.

class faust.agents.Agent(fun: Callable[faust.types.streams.StreamT, Union[Coroutine[[Any, Any], None], Awaitable[None], AsyncIterable]], *, app: faust.types.app.AppT, name: str = None, channel: Union[str, faust.types.channels.ChannelT] = None, concurrency: int = 1, sink: Iterable[Union[AgentT, faust.types.channels.ChannelT, Callable[Any, Optional[Awaitable]]]] = None, on_error: Callable[[AgentT, BaseException], Awaitable] = None, supervisor_strategy: Type[mode.types.supervisors.SupervisorStrategyT] = None, help: str = None, key_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, value_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, isolated_partitions: bool = False, use_reply_headers: bool = None, **kwargs) → None

Agent.

This is the type of object returned by the @app.agent decorator.

supervisor = None

on_init_dependencies() → Iterable[mode.types.services.ServiceT]

Return list of services dependencies required to start agent.

Return type

Iterable[ServiceT[]]

cancel() → None

Cancel agent and its actor instances running in this process.

Return type

None

info() → Mapping

Return agent attributes as a dictionary.

Return type

Mapping[~KT, +VT_co]

clone(*, cls: Type[faust.types.agents.AgentT] = None, **kwargs) → faust.types.agents.AgentT

Create clone of this agent object.

Keyword arguments can be passed to override any argument supported by Agent.__init__.

Return type

AgentT[]

test_context(channel: faust.types.channels.ChannelT = None, supervisor_strategy: mode.types.supervisors.SupervisorStrategyT = None, on_error: Callable[[AgentT, BaseException], Awaitable] = None, **kwargs) → faust.types.agents.AgentTestWrapperT

Create new unit-testing wrapper for this agent.

Return type

AgentTestWrapperT[]

actor_from_stream(stream: Optional[faust.types.streams.StreamT], *, index: int = None, active_partitions: Set[faust.types.tuples.TP] = None, channel: faust.types.channels.ChannelT = None) → faust.types.agents.ActorT[Union[AsyncIterable, Awaitable]]

Create new actor from stream.

Return type

ActorT[]

add_sink(sink: Union[AgentT, faust.types.channels.ChannelT, Callable[Any, Optional[Awaitable]]]) → None

Add new sink to further handle results from this agent.

Return type

None

stream(channel: faust.types.channels.ChannelT = None, active_partitions: Set[faust.types.tuples.TP] = None, **kwargs) → faust.types.streams.StreamT

Create underlying stream used by this agent.

Return type

StreamT[+T_co]

map(values: Union[AsyncIterable, Iterable], key: Union[bytes, faust.types.core._ModelT, Any, None] = None, reply_to: Union[AgentT, faust.types.channels.ChannelT, str] = None) → AsyncIterator

RPC map operation on a list of values.

A map operation iterates over results as they arrive. See join() and kvjoin() if you want them in order.

Return type

AsyncIterator[+T_co]

kvmap(items: Union[AsyncIterable[Tuple[Union[bytes, faust.types.core._ModelT, Any, None], Union[bytes, faust.types.core._ModelT, Any]]], Iterable[Tuple[Union[bytes, faust.types.core._ModelT, Any, None], Union[bytes, faust.types.core._ModelT, Any]]]], reply_to: Union[AgentT, faust.types.channels.ChannelT, str] = None) → AsyncIterator[str]

RPC map operation on a list of (key, value) pairs.

A map operation iterates over results as they arrive. See join() and kvjoin() if you want them in order.

Return type

AsyncIterator[str]

get_topic_names() → Iterable[str]

Return list of topic names this agent subscribes to.

Return type

Iterable[str]

property channel

Return channel used by agent. :rtype: ChannelT[]

property channel_iterator

Return channel agent iterates over. :rtype: AsyncIterator[+T_co]

property label

Return human-readable description of agent. :rtype: str

property shortlabel

Return short description of agent. :rtype: str

logger = <Logger faust.agents.agent (WARNING)>

faust.agents.AgentFun

alias of typing.Callable

class faust.agents.AgentT(fun: Callable[faust.types.streams.StreamT, Union[Coroutine[[Any, Any], None], Awaitable[None], AsyncIterable]], *, name: str = None, app: faust.types.agents._AppT = None, channel: Union[str, faust.types.channels.ChannelT] = None, concurrency: int = 1, sink: Iterable[Union[AgentT, faust.types.channels.ChannelT, Callable[Any, Optional[Awaitable]]]] = None, on_error: Callable[[AgentT, BaseException], Awaitable] = None, supervisor_strategy: Type[mode.types.supervisors.SupervisorStrategyT] = None, help: str = None, key_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, value_type: Union[Type[faust.types.models.ModelT], Type[bytes], Type[str]] = None, isolated_partitions: bool = False, **kwargs) → None

abstract test_context(channel: faust.types.channels.ChannelT = None, supervisor_strategy: mode.types.supervisors.SupervisorStrategyT = None, **kwargs) → faust.types.agents.AgentTestWrapperT

Return type

AgentTestWrapperT[]

abstract add_sink(sink: Union[AgentT, faust.types.channels.ChannelT, Callable[Any, Optional[Awaitable]]]) → None

Return type

None

abstract stream(**kwargs) → faust.types.streams.StreamT

Return type

StreamT[+T_co]

abstract info() → Mapping

Return type

Mapping[~KT, +VT_co]

abstract clone(*, cls: Type[AgentT] = None, **kwargs) → faust.types.agents.AgentT

Return type

AgentT[]

abstract get_topic_names() → Iterable[str]

Return type

Iterable[str]

abstract property channel

Return type

ChannelT[]

abstract property channel_iterator

Return type

AsyncIterator[+T_co]

class faust.agents.AgentManager(app: faust.types.app.AppT, **kwargs) → None

Agent manager.

service_reset() → None

Reset service state on restart.

Return type

None

cancel() → None

Cancel all running agents.

Return type

None

update_topic_index() → None

Update indices.

Return type

None

logger = <Logger faust.agents.manager (WARNING)>

class faust.agents.AgentManagerT(*, beacon: mode.utils.types.trees.NodeT = None, loop: asyncio.events.AbstractEventLoop = None) → None

class faust.agents.ReplyConsumer(app: faust.types.app.AppT, **kwargs) → None

Consumer responsible for redelegation of replies received.

logger = <Logger faust.agents.replies (WARNING)>

faust.agents.current_agent() → Optional[faust.types.agents.AgentT]

Return type

Optional[AgentT[]]