temporalio.plugin.SimplePlugin

class documentation

class SimplePlugin(temporalio.client.Plugin, temporalio.worker.Plugin): (source)

Known subclasses: temporalio.contrib.openai_agents.OpenAIAgentsPlugin, temporalio.contrib.opentelemetry.OpenTelemetryPlugin

Constructor: SimplePlugin(name, data_converter, interceptors, activities, ...)

View In Hierarchy

A simple plugin definition which has a limited set of configurations but makes it easier to produce a plugin which needs to configure them.

Method	`__init__`	component for which they are applicable. All arguments are optional, and all but run_context can also be callables for more complex modification. See the type PluginParameter above. For details on each argument, see below.
Method	`configure_client`	See base class.
Method	`configure_replayer`	See base class.
Method	`configure_worker`	See base class.
Async Method	`connect_service_client`	See base class.
Method	`name`	See base class.
Async Method	`run_replayer`	See base class.
Async Method	`run_worker`	See base class.
Instance Variable	`activities`	Undocumented
Instance Variable	`data_converter`	Undocumented
Instance Variable	`interceptors`	Undocumented
Instance Variable	`nexus_service_handlers`	Undocumented
Instance Variable	`run_context`	Undocumented
Instance Variable	`workflow_failure_exception_types`	Undocumented
Instance Variable	`workflow_runner`	Undocumented
Instance Variable	`workflows`	Undocumented
Instance Variable	`_name`	Undocumented

def __init__(self, name: str, *, data_converter: PluginParameter[temporalio.converter.DataConverter] = None, interceptors: Sequence[temporalio.client.Interceptor | temporalio.worker.Interceptor] | None = None, activities: PluginParameter[Sequence[Callable]] = None, nexus_service_handlers: PluginParameter[Sequence[Any]] = None, workflows: PluginParameter[Sequence[type]] = None, workflow_runner: PluginParameter[WorkflowRunner] = None, workflow_failure_exception_types: PluginParameter[Sequence[type[BaseException]]] = None, run_context: Callable[[], AbstractAsyncContextManager[None]] | None = None): (source) ¶

overridden in temporalio.contrib.openai_agents.OpenAIAgentsPlugin, temporalio.contrib.opentelemetry.OpenTelemetryPlugin

Create a simple plugin with configurable parameters. Each of the parameters will be applied to any: component for which they are applicable. All arguments are optional, and all but run_context can also be callables for more complex modification. See the type PluginParameter above. For details on each argument, see below.

Parameters
name:`str`	The name of the plugin.
data_converter:`PluginParameter[temporalio.converter.DataConverter]`	Data converter for serialization, or callable to customize existing one. Applied to the Client and Replayer.
interceptors:`Sequence[temporalio.client.Interceptor \| temporalio.worker.Interceptor] \| None`	Interceptors to append. Client interceptors are applied to the Client, worker interceptors are applied to the Worker and Replayer. Interceptors that implement both interfaces will be applied to both, with exactly one instance used per worker to avoid duplication.
activities:`PluginParameter[Sequence[Callable]]`	Activity functions to append, or callable to customize existing ones. Applied to the Worker.
nexus_service_handlers:`PluginParameter[Sequence[Any]]`	Nexus service handlers to append, or callable to customize existing ones. Applied to the Worker.
workflows:`PluginParameter[Sequence[type]]`	Workflow classes to append, or callable to customize existing ones. Applied to the Worker and Replayer.
workflow_runner:`PluginParameter[WorkflowRunner]`	Workflow runner, or callable to customize existing one. Applied to the Worker and Replayer.
workflow_failure_exception_types:`PluginParameter[Sequence[type[BaseException]]]`	Exception types for workflow failures to append, or callable to customize existing ones. Applied to the Worker and Replayer.
run_context:`Callable[[], AbstractAsyncContextManager[None]] \| None`	A place to run custom code to wrap around the Worker (or Replayer) execution. Specifically, it's an async context manager producer. Applied to the Worker and Replayer.

Returns
A configured Plugin instance.

def configure_client(self, config: ClientConfig) -> ClientConfig: (source) ¶

overrides temporalio.client.Plugin.configure_client

See base class.

def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig: (source) ¶

overrides temporalio.worker.Plugin.configure_replayer

See base class.

def configure_worker(self, config: WorkerConfig) -> WorkerConfig: (source) ¶

overrides temporalio.worker.Plugin.configure_worker

See base class.

async def connect_service_client(self, config: ConnectConfig, next: Callable[[ConnectConfig], Awaitable[ServiceClient]]) -> temporalio.service.ServiceClient: (source) ¶

overrides temporalio.client.Plugin.connect_service_client

See base class.

def name(self) -> str: (source) ¶

overrides temporalio.client.Plugin.name

See base class.

@asynccontextmanager

async def run_replayer(self, replayer: Replayer, histories: AsyncIterator[WorkflowHistory], next: Callable[[Replayer, AsyncIterator[WorkflowHistory]], AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]]) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]: (source) ¶

overrides temporalio.worker.Plugin.run_replayer

See base class.