temporalio.plugin.SimplePlugin

class documentation

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

Known subclasses: temporalio.contrib.openai_agents.OpenAIAgentsPlugin

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

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	`client_interceptors`	Undocumented
Instance Variable	`data_converter`	Undocumented
Instance Variable	`nexus_service_handlers`	Undocumented
Instance Variable	`run_context`	Undocumented
Instance Variable	`worker_interceptors`	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, client_interceptors: PluginParameter[Sequence[temporalio.client.Interceptor]] = None, activities: PluginParameter[Sequence[Callable]] = None, nexus_service_handlers: PluginParameter[Sequence[Any]] = None, workflows: PluginParameter[Sequence[type]] = None, workflow_runner: PluginParameter[WorkflowRunner] = None, worker_interceptors: PluginParameter[Sequence[temporalio.worker.Interceptor]] = 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

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.
client_interceptors:`PluginParameter[Sequence[temporalio.client.Interceptor]]`	Client interceptors to append, or callable to customize existing ones. Applied to the Client. Note, if the provided interceptor is also a worker.Interceptor, it will be added to any worker which uses that client.
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.
worker_interceptors:`PluginParameter[Sequence[temporalio.worker.Interceptor]]`	Worker interceptors to append, or callable to customize existing ones. 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.