Coverage for async_durable_execution/_core/serdes.py: 94%

1"""Codec-based serialization and deserialization for Python types. 

2 

3This module provides comprehensive serialization support using a codec-based 

4architecture with recursive encoding/decoding for nested structures. 

5 

6Key Features: 

7- Plain JSON for primitives and simple lists (performance optimization) 

8- Envelope format with type tags for complex types 

9- Modular codec architecture 

10- Recursive handling of nested structures 

11 

12Serialization Strategy: 

13- Primitives (None, str, int, float, bool): Plain JSON 

14- Simple lists containing only primitives: Plain JSON 

15- Everything else: Envelope format with type tags 

16 

17Wire Formats: 

18 Plain JSON: 42, "hello", [1, 2, 3] 

19 Envelope: {"t": "<type_tag>", "v": <encoded_value>} 

20""" 

21 

22from __future__ import annotations 

23 

24import base64 

25import json 

26import logging 

27import uuid 

28from abc import ABC, abstractmethod 

29from collections.abc import Callable 

30from dataclasses import dataclass, replace 

31from datetime import date, datetime 

32from decimal import Decimal 

33from enum import Enum 

34from typing import Any, Generic, Literal, Protocol, TypeVar, cast 

35 

36from .context import SerDesContext, bind_current_context, get_current_context 

37from .exceptions import ( 

38 DurableExecutionsError, 

39 ExecutionError, 

40 RetryableSerDesError, 

41 SerDesError, 

42) 

43from .models import OperationSubTypeValue, OperationType 

44 

45logger = logging.getLogger(__name__) 

46 

47T = TypeVar("T") 

48 

49TYPE_TOKEN: str = "t" 

50VALUE_TOKEN: str = "v" 

51 

52 

53class TypeTag(str, Enum): 

54 """Type tags for envelope format.""" 

55 

56 NONE = "n" 

57 STR = "s" 

58 INT = "i" 

59 FLOAT = "f" 

60 BOOL = "b" 

61 BYTES = "B" 

62 UUID = "u" 

63 DECIMAL = "d" 

64 DATETIME = "dt" 

65 DATE = "D" 

66 TUPLE = "t" 

67 LIST = "l" 

68 DICT = "m" 

69 

70 

71@dataclass(frozen=True) 

72class EncodedValue: 

73 """Encoded value with type tag.""" 

74 

75 tag: TypeTag | str 

76 

77 value: Any 

78 

79 

80class Codec(Protocol): 

81 """Protocol for type-specific codecs.""" 

82 

83 def encode(self, obj: Any) -> EncodedValue: ... 

84 

85 def decode(self, tag: TypeTag | str, value: Any) -> Any: ... 

86 

87 

88class TypeCodecExtension(Protocol): 

89 """Extension point for types owned outside the core package.""" 

90 

91 tag: str 

92 

93 def can_encode(self, obj: Any) -> bool: ... 

94 

95 def encode( 

96 self, 

97 obj: Any, 

98 encode_value: Callable[[Any], EncodedValue], 

99 ) -> Any: ... 

100 

101 def decode( 

102 self, 

103 value: Any, 

104 decode_value: Callable[[TypeTag | str, Any], Any], 

105 ) -> Any: ... 

106 

107 

108class PrimitiveCodec: 

109 """Codec for primitive types.""" 

110 

111 def encode(self, obj: Any) -> EncodedValue: # noqa: PLR6301 

112 match obj: 

113 case None: 

114 return EncodedValue(TypeTag.NONE, None) 

115 case str(): 

116 return EncodedValue(TypeTag.STR, obj) 

117 case bool(): # Must come before int 

118 return EncodedValue(TypeTag.BOOL, obj) 

119 case int(): 

120 return EncodedValue(TypeTag.INT, obj) 

121 case float(): 

122 return EncodedValue(TypeTag.FLOAT, obj) 

123 case _: 

124 msg = f"Unsupported primitive type: {type(obj)!r}" 

125 raise SerDesError(msg) 

126 

127 def decode(self, tag: TypeTag, value: Any) -> Any: # noqa: PLR6301 

128 match tag: 

129 case TypeTag.NONE: 

130 return None 

131 case TypeTag.STR: 

132 return str(value) 

133 case TypeTag.BOOL: 

134 return bool(value) 

135 case TypeTag.INT: 

136 return int(value) 

137 case TypeTag.FLOAT: 

138 return float(value) 

139 case _: 

140 msg = f"Unknown primitive tag: {tag}" 

141 raise SerDesError(msg) 

142 

143 

144class BytesCodec: 

145 """Codec for bytes, bytearray, and memoryview.""" 

146 

147 def encode(self, obj: Any) -> EncodedValue: # noqa: PLR6301 

148 encoded = base64.b64encode(bytes(obj)).decode("utf-8") 

149 return EncodedValue(TypeTag.BYTES, encoded) 

150 

151 def decode(self, tag: TypeTag, value: Any) -> Any: # noqa: PLR6301 

152 if tag != TypeTag.BYTES: 

153 msg = f"Expected BYTES tag, got {tag}" 

154 raise SerDesError(msg) 

155 return base64.b64decode(value.encode("utf-8")) 

156 

157 

158class UuidCodec: 

159 """Codec for UUID objects.""" 

160 

161 def encode(self, obj: Any) -> EncodedValue: # noqa: PLR6301 

162 return EncodedValue(TypeTag.UUID, str(obj)) 

163 

164 def decode(self, tag: TypeTag, value: Any) -> Any: # noqa: PLR6301 

165 if tag != TypeTag.UUID: 

166 msg = f"Expected UUID tag, got {tag}" 

167 raise SerDesError(msg) 

168 return uuid.UUID(value) 

169 

170 

171class DecimalCodec: 

172 """Codec for Decimal objects.""" 

173 

174 def encode(self, obj: Any) -> EncodedValue: # noqa: PLR6301 

175 return EncodedValue(TypeTag.DECIMAL, str(obj)) 

176 

177 def decode(self, tag: TypeTag, value: Any) -> Any: # noqa: PLR6301 

178 if tag != TypeTag.DECIMAL: 

179 msg = f"Expected DECIMAL tag, got {tag}" 

180 raise SerDesError(msg) 

181 return Decimal(value) 

182 

183 

184class DateTimeCodec: 

185 """Codec for datetime and date objects.""" 

186 

187 def encode(self, obj: Any) -> EncodedValue: # noqa: PLR6301 

188 match obj: 

189 case datetime(): 

190 return EncodedValue(TypeTag.DATETIME, obj.isoformat()) 

191 case date(): 

192 return EncodedValue(TypeTag.DATE, obj.isoformat()) 

193 case _: 

194 msg = f"Unsupported datetime type: {type(obj)!r}" 

195 raise SerDesError(msg) 

196 

197 def decode(self, tag: TypeTag, value: Any) -> Any: # noqa: PLR6301 

198 match tag: 

199 case TypeTag.DATETIME: 

200 # Handle Z suffix for UTC 

201 s = value 

202 if isinstance(s, str) and s.endswith("Z"): 

203 s = s[:-1] + "+00:00" 

204 return datetime.fromisoformat(s) 

205 case TypeTag.DATE: 

206 return date.fromisoformat(value) 

207 case _: 

208 msg = f"Unknown datetime tag: {tag}" 

209 raise SerDesError(msg) 

210 

211 

212class ContainerCodec(Codec): 

213 """Codec for container types with recursive encoding/decoding.""" 

214 

215 def __init__(self) -> None: 

216 self._dispatcher: TypeCodec | None = None 

217 

218 def set_dispatcher(self, dispatcher) -> None: 

219 """Set the main codec dispatcher for recursive encoding.""" 

220 self._dispatcher = dispatcher 

221 

222 @property 

223 def dispatcher(self) -> TypeCodec: 

224 """Get the dispatcher, raising error if not set.""" 

225 if self._dispatcher is None: 

226 msg = "ContainerCodec not linked to a TypeCodec dispatcher." 

227 raise DurableExecutionsError(msg) 

228 return self._dispatcher 

229 

230 def encode(self, obj: Any) -> EncodedValue: 

231 """Encode container using dispatcher for recursive elements.""" 

232 

233 match obj: 

234 case list(): 

235 return EncodedValue( 

236 TypeTag.LIST, [self._wrap(v, self.dispatcher) for v in obj] 

237 ) 

238 case tuple(): 

239 return EncodedValue( 

240 TypeTag.TUPLE, [self._wrap(v, self.dispatcher) for v in obj] 

241 ) 

242 case dict(): 

243 for k in obj: 

244 if isinstance(k, tuple): 

245 msg = "Tuple keys not supported" 

246 raise SerDesError(msg) 

247 return EncodedValue( 

248 TypeTag.DICT, 

249 {k: self._wrap(v, self.dispatcher) for k, v in obj.items()}, 

250 ) 

251 case _: 

252 msg = f"Unsupported container type: {type(obj)!r}" 

253 raise SerDesError(msg) 

254 

255 def decode(self, tag: TypeTag | str, value: Any) -> Any: 

256 """Decode container using dispatcher for recursive elements.""" 

257 

258 match tag: 

259 case TypeTag.LIST: 

260 if not isinstance(value, list): 

261 msg = f"Expected list, got {type(value)}" 

262 raise SerDesError(msg) 

263 return [self._unwrap(v, self.dispatcher) for v in value] 

264 case TypeTag.TUPLE: 

265 if not isinstance(value, list): 

266 msg = f"Expected list, got {type(value)}" 

267 raise SerDesError(msg) 

268 return tuple(self._unwrap(v, self.dispatcher) for v in value) 

269 case TypeTag.DICT: 

270 if not isinstance(value, dict): 

271 msg = f"Expected dict, got {type(value)}" 

272 raise SerDesError(msg) 

273 return {k: self._unwrap(v, self.dispatcher) for k, v in value.items()} 

274 case _: 

275 msg = f"Unknown container tag: {tag}" 

276 raise SerDesError(msg) 

277 

278 @staticmethod 

279 def _wrap(obj: Any, dispatcher) -> EncodedValue: 

280 """Wrap object using dispatcher.""" 

281 return dispatcher.encode(obj) 

282 

283 @staticmethod 

284 def _unwrap(obj: Any, dispatcher) -> Any: 

285 """Unwrap object using dispatcher.""" 

286 match obj: 

287 case EncodedValue(): 

288 return dispatcher.decode(obj.tag, obj.value) 

289 case dict() if TYPE_TOKEN in obj and VALUE_TOKEN in obj: 

290 return dispatcher.decode(obj[TYPE_TOKEN], obj[VALUE_TOKEN]) 

291 case _: 

292 return obj 

293 

294 

295class TypeCodec(Codec): 

296 """Main codec dispatcher.""" 

297 

298 def __init__( 

299 self, 

300 extensions: tuple[TypeCodecExtension, ...] = (), 

301 ) -> None: 

302 built_in_tags = {tag.value for tag in TypeTag} 

303 extension_tags = [extension.tag for extension in extensions] 

304 if len(extension_tags) != len(set(extension_tags)): 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true

305 msg = "Type codec extension tags must be unique." 

306 raise ValueError(msg) 

307 if built_in_tags.intersection(extension_tags): 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true

308 msg = "Type codec extension tags cannot replace built-in tags." 

309 raise ValueError(msg) 

310 

311 self.extensions = extensions 

312 self.extensions_by_tag = {extension.tag: extension for extension in extensions} 

313 self.primitive_codec = PrimitiveCodec() 

314 self.bytes_codec = BytesCodec() 

315 self.uuid_codec = UuidCodec() 

316 self.decimal_codec = DecimalCodec() 

317 self.datetime_codec = DateTimeCodec() 

318 self.container_codec = ContainerCodec() 

319 self.container_codec.set_dispatcher(self) 

320 

321 def encode(self, obj: Any) -> EncodedValue: 

322 for extension in self.extensions: 

323 if extension.can_encode(obj): 

324 return EncodedValue( 

325 extension.tag, 

326 extension.encode(obj, self.encode), 

327 ) 

328 

329 match obj: 

330 case None | str() | bool() | int() | float(): 

331 return self.primitive_codec.encode(obj) 

332 case bytes() | bytearray() | memoryview(): 

333 return self.bytes_codec.encode(bytes(obj)) 

334 case uuid.UUID(): 

335 return self.uuid_codec.encode(obj) 

336 case Decimal(): 

337 return self.decimal_codec.encode(obj) 

338 case datetime() | date(): 

339 return self.datetime_codec.encode(obj) 

340 case list() | tuple() | dict(): 

341 return self.container_codec.encode(obj) 

342 case _: 

343 msg = f"Unsupported type: {type(obj)}" 

344 raise SerDesError(msg) 

345 

346 def decode(self, tag: TypeTag | str, value: Any) -> Any: 

347 tag_value = tag.value if isinstance(tag, TypeTag) else tag 

348 extension = self.extensions_by_tag.get(tag_value) 

349 if extension is not None: 

350 return extension.decode(value, self.decode) 

351 

352 try: 

353 built_in_tag = TypeTag(tag_value) 

354 except (TypeError, ValueError): 

355 msg = f"Unknown type tag: {tag}" 

356 raise SerDesError(msg) from None 

357 

358 match built_in_tag: 

359 case ( 

360 TypeTag.NONE | TypeTag.STR | TypeTag.BOOL | TypeTag.INT | TypeTag.FLOAT 

361 ): 

362 return self.primitive_codec.decode(built_in_tag, value) 

363 case TypeTag.BYTES: 

364 return self.bytes_codec.decode(built_in_tag, value) 

365 case TypeTag.UUID: 

366 return self.uuid_codec.decode(built_in_tag, value) 

367 case TypeTag.DECIMAL: 

368 return self.decimal_codec.decode(built_in_tag, value) 

369 case TypeTag.DATETIME | TypeTag.DATE: 

370 return self.datetime_codec.decode(built_in_tag, value) 

371 case TypeTag.LIST | TypeTag.TUPLE | TypeTag.DICT: 371 ↛ 373line 371 didn't jump to line 373 because the pattern on line 371 always matched

372 return self.container_codec.decode(built_in_tag, value) 

373 case _: 

374 msg = f"Unknown type tag: {tag}" 

375 raise SerDesError(msg) 

376 

377 def has_tag(self, tag: Any) -> bool: 

378 """Return whether a built-in or extension codec owns a tag.""" 

379 if not isinstance(tag, str): 379 ↛ 380line 379 didn't jump to line 380 because the condition on line 379 was never true

380 return False 

381 return tag in self.extensions_by_tag or tag in {item.value for item in TypeTag} 

382 

383 

384TYPE_CODEC = TypeCodec() 

385 

386 

387def get_serdes_context() -> SerDesContext: 

388 """Return the active `SerDesContext`.""" 

389 current_context = get_current_context() 

390 if not isinstance(current_context, SerDesContext): 

391 msg = ( 

392 "get_serdes_context() can only be used while a SerDes operation " 

393 "is executing." 

394 ) 

395 raise RuntimeError(msg) 

396 return current_context 

397 

398 

399class SerDes(ABC, Generic[T]): 

400 """Abstract serializer interface for durable operation payloads and results.""" 

401 

402 @abstractmethod 

403 async def serialize(self, value: T) -> str: 

404 """Convert a Python value into the wire format stored by the SDK.""" 

405 pass 

406 

407 @abstractmethod 

408 async def deserialize(self, data: str) -> T: 

409 """Reconstruct a Python value from the durable wire format.""" 

410 pass 

411 

412 def then(self, stage: SerDesStage) -> ComposableSerDes[T]: 

413 """Return an immutable pipeline with ``stage`` appended.""" 

414 return ComposableSerDes(self, (stage,)) 

415 

416 @staticmethod 

417 def is_primitive(obj: Any) -> bool: 

418 """Check if object contains only JSON-serializable primitives.""" 

419 if obj is None or isinstance(obj, str | int | float | bool): 

420 return True 

421 if isinstance(obj, list): 

422 return all(SerDes.is_primitive(item) for item in obj) 

423 return False 

424 

425 

426class SerDesStage(Protocol): 

427 """A reversible, self-identifying string transformation. 

428 

429 During deserialization, a stage must reverse recognized valid input, 

430 reject recognized malformed or unsupported input, and return unrecognized 

431 input unchanged. 

432 """ 

433 

434 async def serialize(self, value: str, context: SerDesContext) -> str: 

435 """Transform the preceding pipeline stage's string.""" 

436 ... 

437 

438 async def deserialize(self, data: str, context: SerDesContext) -> str: 

439 """Reverse this stage or return unrecognized input unchanged.""" 

440 ... 

441 

442 

443class SerDesPipelineError(SerDesError): 

444 """Identifies the pipeline component and action that failed.""" 

445 

446 def __init__( 

447 self, 

448 stage_index: int, 

449 action: Literal["serialize", "deserialize"], 

450 stage: object, 

451 ) -> None: 

452 self.stage_index = stage_index 

453 self.action = action 

454 self.stage = stage 

455 component_name = f"{type(stage).__module__}.{type(stage).__qualname__}" 

456 super().__init__( 

457 f"SerDes pipeline stage {stage_index} ({component_name}) failed to {action}" 

458 ) 

459 

460 

461class ComposableSerDes(SerDes[T]): 

462 """Immutable SerDes pipeline with one value codec and string stages.""" 

463 

464 def __init__( 

465 self, 

466 value_codec: SerDes[T], 

467 stages: tuple[SerDesStage, ...] = (), 

468 ) -> None: 

469 if value_codec is None: 469 ↛ 470line 469 didn't jump to line 470 because the condition on line 469 was never true

470 msg = "value_codec must not be None." 

471 raise TypeError(msg) 

472 if any(stage is None for stage in stages): 

473 msg = "pipeline stages must not be None." 

474 raise TypeError(msg) 

475 if isinstance(value_codec, ComposableSerDes): 

476 self._value_codec = value_codec.value_codec 

477 self._stages = (*value_codec.stages, *stages) 

478 else: 

479 self._value_codec = value_codec 

480 self._stages = stages 

481 

482 @property 

483 def value_codec(self) -> SerDes[T]: 

484 """Return the root value codec.""" 

485 return self._value_codec 

486 

487 @property 

488 def stages(self) -> tuple[SerDesStage, ...]: 

489 """Return the ordered immutable string stages.""" 

490 return self._stages 

491 

492 def then(self, stage: SerDesStage) -> ComposableSerDes[T]: 

493 """Return a new pipeline with ``stage`` appended.""" 

494 if stage is None: 494 ↛ 495line 494 didn't jump to line 495 because the condition on line 494 was never true

495 msg = "stage must not be None." 

496 raise TypeError(msg) 

497 return ComposableSerDes(self._value_codec, (*self._stages, stage)) 

498 

499 async def serialize(self, value: T) -> str: 

500 context = _current_or_empty_serdes_context() 

501 stage_context = replace(context, original_value=value) 

502 current = await self._invoke_value_codec_serialize(value) 

503 for index, stage in enumerate(self._stages, start=1): 

504 try: 

505 current = await stage.serialize(current, stage_context) 

506 if not isinstance(current, str): 

507 msg = "Stage returned a non-string value." 

508 raise TypeError(msg) 

509 except RetryableSerDesError as error: 

510 raise RetryableSerDesError( 

511 _pipeline_failure_message(index, "serialize", stage) 

512 ) from error 

513 except Exception as error: 

514 raise SerDesPipelineError(index, "serialize", stage) from error 

515 return current 

516 

517 async def deserialize(self, data: str) -> T: 

518 context = _current_or_empty_serdes_context() 

519 stage_context = replace(context, original_value=None) 

520 current = data 

521 for index in range(len(self._stages) - 1, -1, -1): 

522 stage = self._stages[index] 

523 try: 

524 current = await stage.deserialize(current, stage_context) 

525 if not isinstance(current, str): 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true

526 msg = "Stage returned a non-string value." 

527 raise TypeError(msg) 

528 except RetryableSerDesError as error: 

529 raise RetryableSerDesError( 

530 _pipeline_failure_message(index + 1, "deserialize", stage) 

531 ) from error 

532 except Exception as error: 

533 raise SerDesPipelineError( 

534 index + 1, 

535 "deserialize", 

536 stage, 

537 ) from error 

538 return await self._invoke_value_codec_deserialize(current) 

539 

540 async def _invoke_value_codec_serialize(self, value: T) -> str: 

541 try: 

542 result = await self._value_codec.serialize(value) 

543 if not isinstance(result, str): 543 ↛ 544line 543 didn't jump to line 544 because the condition on line 543 was never true

544 msg = "Value codec returned a non-string value." 

545 raise TypeError(msg) 

546 return result 

547 except RetryableSerDesError as error: 

548 raise RetryableSerDesError( 

549 _pipeline_failure_message(0, "serialize", self._value_codec) 

550 ) from error 

551 except Exception as error: 

552 raise SerDesPipelineError(0, "serialize", self._value_codec) from error 

553 

554 async def _invoke_value_codec_deserialize(self, data: str) -> T: 

555 try: 

556 return await self._value_codec.deserialize(data) 

557 except RetryableSerDesError as error: 

558 raise RetryableSerDesError( 

559 _pipeline_failure_message(0, "deserialize", self._value_codec) 

560 ) from error 

561 except Exception as error: 

562 raise SerDesPipelineError(0, "deserialize", self._value_codec) from error 

563 

564 

565def create_serdes_pipeline( 

566 value_codec: SerDes[T], 

567 *stages: SerDesStage, 

568) -> ComposableSerDes[T]: 

569 """Create an immutable value-codec and string-stage pipeline.""" 

570 if value_codec is None: 

571 msg = "value_codec must not be None." 

572 raise TypeError(msg) 

573 if any(stage is None for stage in stages): 573 ↛ 574line 573 didn't jump to line 574 because the condition on line 573 was never true

574 msg = "pipeline stages must not be None." 

575 raise TypeError(msg) 

576 return ComposableSerDes(value_codec, tuple(stages)) 

577 

578 

579def is_composable_serdes(serdes: SerDes[Any]) -> bool: 

580 """Return whether ``serdes`` is an SDK composable pipeline.""" 

581 return isinstance(serdes, ComposableSerDes) 

582 

583 

584def _current_or_empty_serdes_context() -> SerDesContext: 

585 try: 

586 return get_serdes_context() 

587 except RuntimeError: 

588 return SerDesContext() 

589 

590 

591def _pipeline_failure_message( 

592 stage_index: int, 

593 action: Literal["serialize", "deserialize"], 

594 stage: object, 

595) -> str: 

596 component_name = f"{type(stage).__module__}.{type(stage).__qualname__}" 

597 return f"SerDes pipeline stage {stage_index} ({component_name}) failed to {action}" 

598 

599 

600class PassThroughSerDes(SerDes[T]): 

601 """Serializer that leaves already-serialized string payloads unchanged.""" 

602 

603 async def serialize(self, value: T) -> str: # noqa: PLR6301 

604 return cast("str", value) 

605 

606 async def deserialize(self, data: str) -> T: # noqa: PLR6301 

607 return cast("T", data) 

608 

609 

610class JsonSerDes(SerDes[T]): 

611 """Serializer that uses the standard library `json` module.""" 

612 

613 async def serialize(self, value: T) -> str: # noqa: PLR6301 

614 return json.dumps(value) 

615 

616 async def deserialize(self, data: str) -> T: # noqa: PLR6301 

617 return json.loads(data) 

618 

619 

620class ExtendedTypeSerDes(SerDes[T]): 

621 """Main serializer class.""" 

622 

623 def __init__( 

624 self, 

625 type_codecs: tuple[TypeCodecExtension, ...] = (), 

626 ) -> None: 

627 self._codec = TypeCodec(type_codecs) if type_codecs else TYPE_CODEC 

628 

629 async def serialize(self, value: Any) -> str: 

630 """Serialize value to JSON string.""" 

631 return self.serialize_sync(value) 

632 

633 async def deserialize(self, data: str) -> Any: 

634 """Deserialize JSON string to Python object.""" 

635 return self.deserialize_sync(data) 

636 

637 def serialize_sync(self, value: Any) -> str: 

638 """Serialize value to JSON string without awaiting.""" 

639 # Fast path for primitives 

640 if SerDes.is_primitive(value): 

641 return json.dumps(value, separators=(",", ":")) 

642 

643 self._check_circular_references(value) 

644 encoded = self._codec.encode(value) 

645 wrapped = self._to_json_serializable(encoded) 

646 return json.dumps(wrapped, separators=(",", ":")) 

647 

648 def deserialize_sync(self, data: str) -> Any: 

649 """Deserialize JSON string to Python object without awaiting.""" 

650 obj = json.loads(data) 

651 

652 # Fast path for primitives 

653 if SerDes.is_primitive(obj): 

654 return obj 

655 

656 if not (isinstance(obj, dict) and TYPE_TOKEN in obj and VALUE_TOKEN in obj): 

657 msg = 'Malformed envelope: missing "t" or "v" at root.' 

658 raise SerDesError(msg) 

659 tag = obj[TYPE_TOKEN] 

660 if not self._codec.has_tag(tag): 

661 msg = f'Unknown type tag: "{obj[TYPE_TOKEN]}"' 

662 raise SerDesError(msg) 

663 

664 return self._codec.decode(tag, obj[VALUE_TOKEN]) 

665 

666 def _to_json_serializable(self, obj: Any) -> Any: 

667 """Convert EncodedValue objects to JSON-serializable format.""" 

668 match obj: 

669 case EncodedValue(): 

670 return { 

671 TYPE_TOKEN: ( 

672 obj.tag.value if isinstance(obj.tag, TypeTag) else obj.tag 

673 ), 

674 VALUE_TOKEN: self._to_json_serializable(obj.value), 

675 } 

676 case list(): 

677 return [self._to_json_serializable(x) for x in obj] 

678 case dict(): 

679 return {k: self._to_json_serializable(v) for k, v in obj.items()} 

680 case _: 

681 return obj 

682 

683 def _check_circular_references( 

684 self, obj: Any, seen: set[int] | None = None 

685 ) -> None: 

686 """Reject circular containers before recursive encoding.""" 

687 if not isinstance(obj, (dict, list, tuple)): 

688 return 

689 

690 if seen is None: 

691 seen = set() 

692 

693 obj_id = id(obj) 

694 if obj_id in seen: 

695 msg = "Circular references are not supported" 

696 raise SerDesError(msg) 

697 

698 seen.add(obj_id) 

699 try: 

700 values = obj.values() if isinstance(obj, dict) else obj 

701 for value in values: 

702 self._check_circular_references(value, seen) 

703 finally: 

704 seen.remove(obj_id) 

705 

706 

707DEFAULT_JSON_SERDES: SerDes[Any] = JsonSerDes() 

708EXTENDED_TYPES_SERDES: SerDes[Any] = ExtendedTypeSerDes() 

709 

710 

711async def serialize( 

712 serdes: SerDes[T] | None, 

713 value: T, 

714 operation_id: str, 

715 durable_execution_arn: str, 

716 recursive_level: int = 0, 

717 *, 

718 entity_id: str | None = None, 

719 operation_name: str | None = None, 

720 parent_id: str | None = None, 

721 operation_type: OperationType | None = None, 

722 operation_sub_type: OperationSubTypeValue | None = None, 

723 attempt: int | None = None, 

724) -> str: 

725 """Serialize value using provided or default serializer. 

726 

727 Args: 

728 serdes: Custom serializer or None for default 

729 value: Object to serialize 

730 operation_id: Unique operation identifier 

731 durable_execution_arn: ARN of durable execution 

732 

733 Returns: 

734 Serialized string representation 

735 

736 Raises: 

737 FatalError: If serialization fails 

738 """ 

739 serdes_context: SerDesContext = SerDesContext( 

740 operation_id, 

741 durable_execution_arn, 

742 recursive_level, 

743 entity_id=entity_id or f"operation/{operation_id}", 

744 operation_name=operation_name, 

745 parent_id=parent_id, 

746 operation_type=operation_type, 

747 operation_sub_type=operation_sub_type, 

748 attempt=attempt, 

749 ) 

750 active_serdes: SerDes[T] = serdes or EXTENDED_TYPES_SERDES 

751 

752 async def serialize_value() -> str: 

753 return await active_serdes.serialize(value) 

754 

755 try: 

756 with bind_current_context(serdes_context): 

757 return await serialize_value() 

758 except RetryableSerDesError: 

759 logger.exception( 

760 "⚠️ Retryable serialization failure for id: %s", 

761 operation_id, 

762 ) 

763 raise 

764 except Exception as e: 

765 logger.exception( 

766 "⚠️ Serialization failed for id: %s", 

767 operation_id, 

768 ) 

769 msg = f"Serialization failed for id: {operation_id}, error: {e}." 

770 raise ExecutionError(msg) from e 

771 

772 

773async def deserialize( 

774 serdes: SerDes[T] | None, 

775 data: str, 

776 operation_id: str, 

777 durable_execution_arn: str, 

778 recursive_level: int = 0, 

779 *, 

780 entity_id: str | None = None, 

781 operation_name: str | None = None, 

782 parent_id: str | None = None, 

783 operation_type: OperationType | None = None, 

784 operation_sub_type: OperationSubTypeValue | None = None, 

785 attempt: int | None = None, 

786) -> T: 

787 """Deserialize data using provided or default serializer. 

788 

789 Args: 

790 serdes: Custom serializer or None for default 

791 data: Serialized string data 

792 operation_id: Unique operation identifier 

793 durable_execution_arn: ARN of durable execution 

794 

795 Returns: 

796 Deserialized Python object 

797 

798 Raises: 

799 FatalError: If deserialization fails 

800 """ 

801 serdes_context: SerDesContext = SerDesContext( 

802 operation_id, 

803 durable_execution_arn, 

804 recursive_level, 

805 entity_id=entity_id or f"operation/{operation_id}", 

806 operation_name=operation_name, 

807 parent_id=parent_id, 

808 operation_type=operation_type, 

809 operation_sub_type=operation_sub_type, 

810 attempt=attempt, 

811 ) 

812 active_serdes: SerDes[T] = serdes or EXTENDED_TYPES_SERDES 

813 

814 async def deserialize_value() -> T: