0
# Common Utilities and Configuration
1

2
Core utility classes, enums, and configuration objects used throughout the Temporal Python SDK. These classes provide fundamental building blocks for retry policies, workflow configuration, search attributes, metrics, and other common functionality.
3

4
## Capabilities
5

6
### Retry Configuration
7

8
Configure retry behavior for workflows and activities with exponential backoff and custom error handling.
9

10
```python { .api }
11
@dataclass
12
class RetryPolicy:
13
    initial_interval: timedelta = timedelta(seconds=1)
14
    backoff_coefficient: float = 2.0
15
    maximum_interval: Optional[timedelta] = None
16
    maximum_attempts: int = 0
17
    non_retryable_error_types: Optional[Sequence[str]] = None
18

19
    @staticmethod
20
    def from_proto(proto: temporalio.api.common.v1.RetryPolicy) -> RetryPolicy: ...
21

22
    def apply_to_proto(self, proto: temporalio.api.common.v1.RetryPolicy) -> None: ...
23
```
24

25
#### Usage Examples
26

27
```python
28
from temporalio.common import RetryPolicy
29
from datetime import timedelta
30

31
# Basic retry policy with exponential backoff
32
retry_policy = RetryPolicy(
33
    initial_interval=timedelta(seconds=1),
34
    backoff_coefficient=2.0,
35
    maximum_interval=timedelta(minutes=10),
36
    maximum_attempts=5,
37
    non_retryable_error_types=["ValueError", "TypeError"]
38
)
39

40
# Use in workflow execution
41
result = await client.execute_workflow(
42
    MyWorkflow.run,
43
    "arg",
44
    id="workflow-id",
45
    task_queue="my-queue",
46
    retry_policy=retry_policy
47
)
48
```
49

50
### Workflow ID Policies
51

52
Control how workflow IDs are handled when starting workflows with potentially conflicting identifiers.
53

54
```python { .api }
55
class WorkflowIDReusePolicy(IntEnum):
56
    ALLOW_DUPLICATE = 0
57
    ALLOW_DUPLICATE_FAILED_ONLY = 1
58
    REJECT_DUPLICATE = 2
59
    TERMINATE_IF_RUNNING = 3
60

61
class WorkflowIDConflictPolicy(IntEnum):
62
    UNSPECIFIED = 0
63
    FAIL = 1
64
    USE_EXISTING = 2
65
    TERMINATE_EXISTING = 3
66
```
67

68
### Search Attributes
69

70
Type-safe search attribute handling with strongly typed keys and values for workflow search and filtering.
71

72
```python { .api }
73
class SearchAttributeKey(ABC, Generic[SearchAttributeValueType]):
74
    @property
75
    @abstractmethod
76
    def name(self) -> str: ...
77

78
    @property
79
    @abstractmethod
80
    def indexed_value_type(self) -> SearchAttributeIndexedValueType: ...
81

82
    @property
83
    @abstractmethod
84
    def value_type(self) -> Type[SearchAttributeValueType]: ...
85

86
    @staticmethod
87
    def for_text(name: str) -> SearchAttributeKey[str]: ...
88

89
    @staticmethod
90
    def for_keyword(name: str) -> SearchAttributeKey[str]: ...
91

92
    @staticmethod
93
    def for_int(name: str) -> SearchAttributeKey[int]: ...
94

95
    @staticmethod
96
    def for_float(name: str) -> SearchAttributeKey[float]: ...
97

98
    @staticmethod
99
    def for_bool(name: str) -> SearchAttributeKey[bool]: ...
100

101
    @staticmethod
102
    def for_datetime(name: str) -> SearchAttributeKey[datetime]: ...
103

104
    @staticmethod
105
    def for_keyword_list(name: str) -> SearchAttributeKey[Sequence[str]]: ...
106

107
@dataclass(frozen=True)
108
class TypedSearchAttributes(Collection[SearchAttributePair]):
109
    search_attributes: Sequence[SearchAttributePair]
110

111
    def updated(self, *search_attributes: SearchAttributePair) -> TypedSearchAttributes: ...
112

113
class SearchAttributeUpdate(ABC, Generic[SearchAttributeValueType]):
114
    @property
115
    @abstractmethod
116
    def key(self) -> SearchAttributeKey[SearchAttributeValueType]: ...
117

118
    @property
119
    @abstractmethod
120
    def value(self) -> Optional[SearchAttributeValueType]: ...
121
```
122

123
#### Usage Examples
124

125
```python
126
from temporalio.common import SearchAttributeKey, TypedSearchAttributes
127
from datetime import datetime
128

129
# Create typed search attribute keys
130
customer_id = SearchAttributeKey.for_keyword("CustomerId")
131
order_total = SearchAttributeKey.for_float("OrderTotal")
132
created_at = SearchAttributeKey.for_datetime("CreatedAt")
133

134
# Create search attributes
135
search_attributes = TypedSearchAttributes([
136
    (customer_id, "customer-123"),
137
    (order_total, 99.99),
138
    (created_at, datetime.now())
139
])
140

141
# Use in workflow execution
142
result = await client.execute_workflow(
143
    OrderWorkflow.run,
144
    order_data,
145
    id="order-workflow",
146
    task_queue="orders",
147
    search_attributes=search_attributes
148
)
149
```
150

151
### Metrics
152

153
Comprehensive metrics system for monitoring workflows, activities, and SDK performance.
154

155
```python { .api }
156
class MetricMeter(ABC):
157
    noop: ClassVar[MetricMeter]
158

159
    @abstractmethod
160
    def create_counter(
161
        self, name: str, description: Optional[str] = None, unit: Optional[str] = None
162
    ) -> MetricCounter: ...
163

164
    @abstractmethod
165
    def create_histogram(
166
        self, name: str, description: Optional[str] = None, unit: Optional[str] = None
167
    ) -> MetricHistogram: ...
168

169
    @abstractmethod
170
    def create_histogram_float(
171
        self, name: str, description: Optional[str] = None, unit: Optional[str] = None
172
    ) -> MetricHistogramFloat: ...
173

174
    @abstractmethod
175
    def create_histogram_timedelta(
176
        self, name: str, description: Optional[str] = None, unit: Optional[str] = None
177
    ) -> MetricHistogramTimedelta: ...
178

179
    @abstractmethod
180
    def create_gauge(
181
        self, name: str, description: Optional[str] = None, unit: Optional[str] = None
182
    ) -> MetricGauge: ...
183

184
    @abstractmethod
185
    def create_gauge_float(
186
        self, name: str, description: Optional[str] = None, unit: Optional[str] = None
187
    ) -> MetricGaugeFloat: ...
188

189
class MetricCounter(MetricCommon):
190
    @abstractmethod
191
    def add(
192
        self, value: int, additional_attributes: Optional[MetricAttributes] = None
193
    ) -> None: ...
194

195
class MetricHistogram(MetricCommon):
196
    @abstractmethod
197
    def record(
198
        self, value: int, additional_attributes: Optional[MetricAttributes] = None
199
    ) -> None: ...
200

201
class MetricGauge(MetricCommon):
202
    @abstractmethod
203
    def set(
204
        self, value: int, additional_attributes: Optional[MetricAttributes] = None
205
    ) -> None: ...
206
```
207

208
### Task Priority
209

210
Control task scheduling priority and fairness for workflows and activities.
211

212
```python { .api }
213
@dataclass(frozen=True)
214
class Priority:
215
    priority_key: Optional[int] = None
216
    fairness_key: Optional[str] = None
217
    fairness_weight: Optional[float] = None
218

219
    default: ClassVar[Priority]
220
```
221

222
#### Usage Examples
223

224
```python
225
from temporalio.common import Priority
226

227
# High priority workflow
228
high_priority = Priority(priority_key=1)
229

230
# Tenant-based fairness
231
tenant_priority = Priority(
232
    priority_key=3,
233
    fairness_key="tenant-abc",
234
    fairness_weight=1.5
235
)
236
```
237

238
### Raw Payload Handling
239

240
Direct access to unconverted payload data for advanced use cases.
241

242
```python { .api }
243
@dataclass(frozen=True)
244
class RawValue:
245
    payload: temporalio.api.common.v1.Payload
246
```
247

248
### Versioning
249

250
Control worker versioning behavior for gradual deployments and rollbacks.
251

252
```python { .api }
253
class VersioningBehavior(IntEnum):
254
    UNSPECIFIED = 0
255
    PINNED = 1
256
    AUTO_UPGRADE = 2
257

258
@dataclass(frozen=True)
259
class WorkerDeploymentVersion:
260
    deployment_name: str
261
    build_id: str
262

263
    def to_canonical_string(self) -> str: ...
264

265
    @staticmethod
266
    def from_canonical_string(canonical: str) -> WorkerDeploymentVersion: ...
267

268
class VersioningOverride(ABC):
269
    @abstractmethod
270
    def _to_proto(self) -> temporalio.api.workflow.v1.VersioningOverride: ...
271

272
@dataclass(frozen=True)
273
class PinnedVersioningOverride(VersioningOverride):
274
    version: WorkerDeploymentVersion
275

276
@dataclass(frozen=True)
277
class AutoUpgradeVersioningOverride(VersioningOverride):
278
    pass
279
```
280

281
### Query Configuration
282

283
Configure when queries should be rejected based on workflow state.
284

285
```python { .api }
286
class QueryRejectCondition(IntEnum):
287
    NONE = 0
288
    NOT_OPEN = 1
289
    NOT_COMPLETED_CLEANLY = 2
290
```
291

292
### Header Encoding
293

294
Control automatic header encoding and decoding behavior.
295

296
```python { .api }
297
class HeaderCodecBehavior(IntEnum):
298
    NO_CODEC = 1
299
    CODEC = 2
300
    WORKFLOW_ONLY_CODEC = 3
301
```
302

303
## Types
304

305
```python { .api }
306
from typing import TypeVar, Generic, Sequence, Mapping, List, Union, Optional
307
from datetime import datetime, timedelta
308

309
# Search attribute types
310
SearchAttributeValues = Union[
311
    List[str], List[int], List[float], List[bool], List[datetime]
312
]
313
SearchAttributes = Mapping[str, SearchAttributeValues]
314
SearchAttributeValue = Union[str, int, float, bool, datetime, Sequence[str]]
315
SearchAttributeValueType = TypeVar("SearchAttributeValueType", str, int, float, bool, datetime, Sequence[str])
316

317
# Metrics types
318
MetricAttributes = Mapping[str, Union[str, int, float, bool]]
319

320
class SearchAttributeIndexedValueType(IntEnum):
321
    TEXT = 0
322
    KEYWORD = 1
323
    INT = 2
324
    DOUBLE = 3
325
    BOOL = 4
326
    DATETIME = 5
327
    KEYWORD_LIST = 6
328
```