Skip to content

Initial rough framework for plugins #952

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 8 commits into
base: main
Choose a base branch
from
Draft

Initial rough framework for plugins #952

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
defe0bb
Initial rough framework for plugins
tconley1428 Jul 9, 2025
66b031c
Ensure plugin modification happen before any other initialization
tconley1428 Jul 10, 2025
381e0cc
Format
tconley1428 Jul 10, 2025
0db686d
Use local tuner for type inference
tconley1428 Jul 10, 2025
6bcd2fd
Format
tconley1428 Jul 11, 2025
b5db959
Some PR updates, refactoring tests, added name
tconley1428 Jul 11, 2025
759e650
Added docstrings
tconley1428 Jul 11, 2025
e1dfe5c
Update readme
tconley1428 Jul 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 49 additions & 7 deletions temporalio/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ async def connect(
namespace: str = "default",
api_key: Optional[str] = None,
data_converter: temporalio.converter.DataConverter = temporalio.converter.DataConverter.default,
plugins: Sequence[Plugin] = [],
interceptors: Sequence[Interceptor] = [],
default_workflow_query_reject_condition: Optional[
temporalio.common.QueryRejectCondition
Expand Down Expand Up @@ -178,13 +179,21 @@ async def connect(
runtime=runtime,
http_connect_proxy_config=http_connect_proxy_config,
)

root_plugin: Plugin = _RootPlugin()
for plugin in reversed(list(plugins)):
root_plugin = plugin.init_client_plugin(root_plugin)

service_client = await root_plugin.connect_service_client(connect_config)

return Client(
await temporalio.service.ServiceClient.connect(connect_config),
service_client,
namespace=namespace,
data_converter=data_converter,
interceptors=interceptors,
default_workflow_query_reject_condition=default_workflow_query_reject_condition,
header_codec_behavior=header_codec_behavior,
plugins=plugins,
)

def __init__(
Expand All @@ -193,6 +202,7 @@ def __init__(
*,
namespace: str = "default",
data_converter: temporalio.converter.DataConverter = temporalio.converter.DataConverter.default,
plugins: Sequence[Plugin] = [],
interceptors: Sequence[Interceptor] = [],
default_workflow_query_reject_condition: Optional[
temporalio.common.QueryRejectCondition
Expand All @@ -203,21 +213,28 @@ def __init__(

See :py:meth:`connect` for details on the parameters.
"""
# Iterate over interceptors in reverse building the impl
self._impl: OutboundInterceptor = _ClientImpl(self)
for interceptor in reversed(list(interceptors)):
self._impl = interceptor.intercept_client(self._impl)

# Store the config for tracking
self._config = ClientConfig(
config = ClientConfig(
service_client=service_client,
namespace=namespace,
data_converter=data_converter,
interceptors=interceptors,
default_workflow_query_reject_condition=default_workflow_query_reject_condition,
header_codec_behavior=header_codec_behavior,
plugins=plugins,
)

root_plugin: Plugin = _RootPlugin()
for plugin in reversed(list(plugins)):
root_plugin = plugin.init_client_plugin(root_plugin)

self._config = root_plugin.on_create_client(config)

# Iterate over interceptors in reverse building the impl
self._impl: OutboundInterceptor = _ClientImpl(self)
for interceptor in reversed(list(interceptors)):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This interceptor list now needs to be pulled off the self._config. Don't use any more parameters after on_create_client is called.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good call. I also need to go through and make a good set of tests around this sort of thing.

self._impl = interceptor.intercept_client(self._impl)

def config(self) -> ClientConfig:
"""Config, as a dictionary, used to create this client.

Expand Down Expand Up @@ -1510,6 +1527,7 @@ class ClientConfig(TypedDict, total=False):
Optional[temporalio.common.QueryRejectCondition]
]
header_codec_behavior: Required[HeaderCodecBehavior]
plugins: Required[Sequence[Plugin]]


class WorkflowHistoryEventFilterType(IntEnum):
Expand Down Expand Up @@ -7367,3 +7385,27 @@ async def _decode_user_metadata(
if not metadata.HasField("details")
else (await converter.decode([metadata.details]))[0],
)


class Plugin:
def init_client_plugin(self, next: Plugin) -> Plugin:
self.next_client_plugin = next
return self

def on_create_client(self, config: ClientConfig) -> ClientConfig:
return self.next_client_plugin.on_create_client(config)

async def connect_service_client(
self, config: temporalio.service.ConnectConfig
) -> temporalio.service.ServiceClient:
return await self.next_client_plugin.connect_service_client(config)


class _RootPlugin(Plugin):
def on_create_client(self, config: ClientConfig) -> ClientConfig:
return config

async def connect_service_client(
self, config: temporalio.service.ConnectConfig
) -> temporalio.service.ServiceClient:
return await temporalio.service.ServiceClient.connect(config)
2 changes: 2 additions & 0 deletions temporalio/worker/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
WorkflowSlotInfo,
)
from ._worker import (
Plugin,
PollerBehavior,
PollerBehaviorAutoscaling,
PollerBehaviorSimpleMaximum,
Expand Down Expand Up @@ -78,6 +79,7 @@
"ActivityOutboundInterceptor",
"WorkflowInboundInterceptor",
"WorkflowOutboundInterceptor",
"Plugin",
# Interceptor input
"ContinueAsNewInput",
"ExecuteActivityInput",
Expand Down
Loading
Loading