Coverage for async_durable_execution/_core/aws_http.py: 83%

1"""Model-free AWS Lambda HTTP clients. 

2 

3The clients in this module deliberately bypass botocore's generated Lambda 

4service model. Botocore remains responsible for credential discovery, endpoint 

5metadata, SigV4 signing, and the synchronous HTTP transport. The optional 

6``httpx`` extra supplies HTTPX as the asynchronous transport. 

7""" 

8 

9from __future__ import annotations 

10 

11import asyncio 

12import datetime 

13import importlib 

14import json 

15import random 

16import time 

17from collections.abc import Mapping 

18from dataclasses import dataclass 

19from typing import Any, Protocol, cast 

20from urllib.parse import quote, urlencode, urlsplit 

21from uuid import uuid4 

22 

23from botocore.auth import SigV4Auth 

24from botocore.awsrequest import AWSPreparedRequest, AWSRequest 

25from botocore.config import Config 

26from botocore.configprovider import ConfiguredEndpointProvider 

27from botocore.exceptions import ( 

28 ClientError, 

29 InvalidRetryConfigurationError, 

30 InvalidRetryModeError, 

31 NoCredentialsError, 

32 NoRegionError, 

33 UnknownEndpointError, 

34) 

35from botocore.httpsession import URLLib3Session 

36from botocore.regions import EndpointResolver 

37from botocore.session import Session, get_session 

38from botocore.utils import ensure_boolean 

39from botocore.utils import get_environ_proxies 

40 

41from ..__about__ import __version__ 

42 

43 

44_LAMBDA_SERVICE_NAME = "lambda" 

45_DURABLE_API_VERSION = "2025-12-01" 

46_INVOKE_API_VERSION = "2015-03-31" 

47_DEFAULT_CONNECT_TIMEOUT_SECONDS = 5 

48_DEFAULT_READ_TIMEOUT_SECONDS = 50 

49_DEFAULT_MAX_POOL_CONNECTIONS = 10 

50_DEFAULT_LEGACY_MAX_ATTEMPTS = 5 

51_DEFAULT_STANDARD_MAX_ATTEMPTS = 3 

52_MAX_RETRY_DELAY_SECONDS = 20.0 

53_RETRYABLE_STATUS_CODES = frozenset({429, 500, 502, 503, 504, 509}) 

54_RETRYABLE_ERROR_CODES = frozenset( 

55 { 

56 "Throttling", 

57 "ThrottlingException", 

58 "ThrottledException", 

59 "RequestThrottledException", 

60 "ProvisionedThroughputExceededException", 

61 } 

62) 

63 

64 

65class _SyncHttpSession(Protocol): 

66 def send(self, request: AWSPreparedRequest) -> Any: ... 

67 

68 def close(self) -> None: ... 

69 

70 

71class _AsyncHttpClient(Protocol): 

72 async def request( 

73 self, 

74 method: str, 

75 url: str, 

76 *, 

77 content: bytes, 

78 headers: Mapping[str, str], 

79 ) -> Any: ... 

80 

81 async def aclose(self) -> None: ... 

82 

83 

84@dataclass(frozen=True) 

85class _RequestSpec: 

86 method: str 

87 path: str 

88 query: Mapping[str, Any] 

89 headers: Mapping[str, str] 

90 body: bytes 

91 

92 

93@dataclass(frozen=True) 

94class _EndpointResolution: 

95 endpoint_url: str 

96 signing_region: str 

97 signing_name: str 

98 

99 

100def _quote_path_value(value: Any) -> str: 

101 return quote(str(value), safe="-_.~") 

102 

103 

104def _query_value(value: Any) -> str: 

105 if isinstance(value, bool): 

106 return "true" if value else "false" 

107 return str(value) 

108 

109 

110def _json_default(value: Any) -> Any: 

111 if isinstance(value, datetime.datetime): 111 ↛ 113line 111 didn't jump to line 113 because the condition on line 111 was always true

112 return value.timestamp() 

113 if isinstance(value, datetime.date): 

114 return datetime.datetime.combine( 

115 value, 

116 datetime.time(), 

117 tzinfo=datetime.timezone.utc, 

118 ).timestamp() 

119 if isinstance(value, bytes): 

120 return value.decode("utf-8") 

121 msg = f"Object of type {type(value).__name__} is not JSON serializable" 

122 raise TypeError(msg) 

123 

124 

125def _json_bytes(value: Any) -> bytes: 

126 return json.dumps( 

127 value, 

128 default=_json_default, 

129 ensure_ascii=False, 

130 separators=(",", ":"), 

131 ).encode("utf-8") 

132 

133 

134def _body_params( 

135 params: Mapping[str, Any], 

136 *, 

137 excluded: frozenset[str], 

138) -> dict[str, Any]: 

139 return { 

140 key: value 

141 for key, value in params.items() 

142 if key not in excluded and value is not None 

143 } 

144 

145 

146def _request_spec(operation_name: str, params: Mapping[str, Any]) -> _RequestSpec: 

147 if operation_name == "CheckpointDurableExecution": 

148 execution_arn = _quote_path_value(params["DurableExecutionArn"]) 

149 return _RequestSpec( 

150 method="POST", 

151 path=( 

152 f"/{_DURABLE_API_VERSION}/durable-executions/{execution_arn}/checkpoint" 

153 ), 

154 query={}, 

155 headers={"Content-Type": "application/json"}, 

156 body=_json_bytes( 

157 _body_params( 

158 params, 

159 excluded=frozenset({"DurableExecutionArn"}), 

160 ) 

161 ), 

162 ) 

163 

164 if operation_name == "GetDurableExecutionState": 

165 execution_arn = _quote_path_value(params["DurableExecutionArn"]) 

166 return _RequestSpec( 

167 method="GET", 

168 path=(f"/{_DURABLE_API_VERSION}/durable-executions/{execution_arn}/state"), 

169 query=_body_params( 

170 params, 

171 excluded=frozenset({"DurableExecutionArn"}), 

172 ), 

173 headers={}, 

174 body=b"", 

175 ) 

176 

177 if operation_name == "Invoke": 

178 function_name = _quote_path_value(params["FunctionName"]) 

179 headers = {} 

180 header_names = { 

181 "InvocationType": "X-Amz-Invocation-Type", 

182 "LogType": "X-Amz-Log-Type", 

183 "ClientContext": "X-Amz-Client-Context", 

184 "DurableExecutionName": "X-Amz-Durable-Execution-Name", 

185 "TenantId": "X-Amz-Tenant-Id", 

186 } 

187 for parameter_name, header_name in header_names.items(): 

188 value = params.get(parameter_name) 

189 if value is not None: 

190 headers[header_name] = str(value) 

191 query = {} 

192 if params.get("Qualifier") is not None: 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true

193 query["Qualifier"] = params["Qualifier"] 

194 payload = params.get("Payload", b"") 

195 if isinstance(payload, str): 195 ↛ 196line 195 didn't jump to line 196 because the condition on line 195 was never true

196 payload = payload.encode("utf-8") 

197 return _RequestSpec( 

198 method="POST", 

199 path=f"/{_INVOKE_API_VERSION}/functions/{function_name}/invocations", 

200 query=query, 

201 headers=headers, 

202 body=cast(bytes, payload), 

203 ) 

204 

205 if operation_name == "GetDurableExecution": 

206 execution_arn = _quote_path_value(params["DurableExecutionArn"]) 

207 return _RequestSpec( 

208 method="GET", 

209 path=f"/{_DURABLE_API_VERSION}/durable-executions/{execution_arn}", 

210 query=_body_params( 

211 params, 

212 excluded=frozenset({"DurableExecutionArn"}), 

213 ), 

214 headers={}, 

215 body=b"", 

216 ) 

217 

218 if operation_name == "GetDurableExecutionHistory": 

219 execution_arn = _quote_path_value(params["DurableExecutionArn"]) 

220 return _RequestSpec( 

221 method="GET", 

222 path=( 

223 f"/{_DURABLE_API_VERSION}/durable-executions/{execution_arn}/history" 

224 ), 

225 query=_body_params( 

226 params, 

227 excluded=frozenset({"DurableExecutionArn"}), 

228 ), 

229 headers={}, 

230 body=b"", 

231 ) 

232 

233 if operation_name == "SendDurableExecutionCallbackSuccess": 

234 callback_id = _quote_path_value(params["CallbackId"]) 

235 result = params.get("Result", b"") 

236 if result is None: 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true

237 result = b"" 

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

239 result = result.encode("utf-8") 

240 return _RequestSpec( 

241 method="POST", 

242 path=( 

243 f"/{_DURABLE_API_VERSION}/durable-execution-callbacks/" 

244 f"{callback_id}/succeed" 

245 ), 

246 query={}, 

247 headers={}, 

248 body=cast(bytes, result), 

249 ) 

250 

251 if operation_name == "SendDurableExecutionCallbackFailure": 

252 callback_id = _quote_path_value(params["CallbackId"]) 

253 error = params.get("Error") or {} 

254 return _RequestSpec( 

255 method="POST", 

256 path=( 

257 f"/{_DURABLE_API_VERSION}/durable-execution-callbacks/" 

258 f"{callback_id}/fail" 

259 ), 

260 query={}, 

261 headers={"Content-Type": "application/json"}, 

262 body=_json_bytes(error), 

263 ) 

264 

265 if operation_name == "SendDurableExecutionCallbackHeartbeat": 265 ↛ 278line 265 didn't jump to line 278 because the condition on line 265 was always true

266 callback_id = _quote_path_value(params["CallbackId"]) 

267 return _RequestSpec( 

268 method="POST", 

269 path=( 

270 f"/{_DURABLE_API_VERSION}/durable-execution-callbacks/" 

271 f"{callback_id}/heartbeat" 

272 ), 

273 query={}, 

274 headers={}, 

275 body=b"", 

276 ) 

277 

278 msg = f"Unsupported Lambda operation: {operation_name}" 

279 raise ValueError(msg) 

280 

281 

282def _configured_endpoint_url( 

283 session: Session, 

284 *, 

285 ignore_configured_endpoint_urls: bool | None = None, 

286) -> str | None: 

287 ignore_configured = ignore_configured_endpoint_urls 

288 if ignore_configured is None: 

289 ignore_configured = ensure_boolean( 

290 session.get_config_variable("ignore_configured_endpoint_urls") 

291 ) 

292 if ignore_configured: 

293 return None 

294 provider = ConfiguredEndpointProvider( 

295 full_config=session.full_config, 

296 scoped_config=session.get_scoped_config(), 

297 client_name=_LAMBDA_SERVICE_NAME, 

298 ) 

299 return cast(str | None, provider.provide()) 

300 

301 

302def _resolve_region(session: Session, region_name: str | None) -> str: 

303 resolved_region = region_name or session.get_config_variable("region") 

304 if not resolved_region: 

305 raise NoRegionError() 

306 return cast(str, resolved_region) 

307 

308 

309def _normalize_fips_region(region_name: str) -> tuple[str, bool]: 

310 if region_name.startswith("fips-"): 

311 return region_name.removeprefix("fips-"), True 

312 if region_name.endswith("-fips"): 

313 return region_name.removesuffix("-fips"), True 

314 return region_name, False 

315 

316 

317def _resolve_endpoint( 

318 session: Session, 

319 *, 

320 region_name: str, 

321 endpoint_url: str | None, 

322 use_dualstack_endpoint: bool | None = None, 

323 use_fips_endpoint: bool | None = None, 

324 ignore_configured_endpoint_urls: bool | None = None, 

325) -> _EndpointResolution: 

326 resolved_endpoint = endpoint_url or _configured_endpoint_url( 

327 session, 

328 ignore_configured_endpoint_urls=ignore_configured_endpoint_urls, 

329 ) 

330 

331 if use_dualstack_endpoint is None: 

332 use_dualstack_endpoint = ensure_boolean( 

333 session.get_config_variable("use_dualstack_endpoint") 

334 ) 

335 if use_fips_endpoint is None: 

336 use_fips_endpoint = ensure_boolean( 

337 session.get_config_variable("use_fips_endpoint") 

338 ) 

339 endpoint = EndpointResolver(session.get_data("endpoints")).construct_endpoint( 

340 _LAMBDA_SERVICE_NAME, 

341 region_name, 

342 use_dualstack_endpoint=use_dualstack_endpoint, 

343 use_fips_endpoint=use_fips_endpoint, 

344 ) 

345 if endpoint is None and not resolved_endpoint: 

346 raise UnknownEndpointError( 

347 service_name=_LAMBDA_SERVICE_NAME, 

348 region_name=region_name, 

349 ) 

350 

351 credential_scope = endpoint.get("credentialScope", {}) if endpoint else {} 

352 signing_region = credential_scope.get("region", region_name) 

353 signing_name = credential_scope.get("service", _LAMBDA_SERVICE_NAME) 

354 if resolved_endpoint: 

355 resolved_url = resolved_endpoint.rstrip("/") 

356 else: 

357 assert endpoint is not None 

358 protocols = endpoint.get("protocols") or ["https"] 

359 resolved_url = f"{protocols[0]}://{endpoint['hostname']}" 

360 

361 return _EndpointResolution( 

362 endpoint_url=resolved_url, 

363 signing_region=signing_region, 

364 signing_name=signing_name, 

365 ) 

366 

367 

368def _resolve_max_attempts(session: Session, config: Config) -> int: 

369 config_values = cast("Any", config) 

370 configured_retries = config_values.retries or {} 

371 

372 retry_mode = ( 

373 configured_retries.get("mode") 

374 or session.get_config_variable("retry_mode") 

375 or "legacy" 

376 ) 

377 if retry_mode not in {"legacy", "standard", "adaptive"}: 

378 raise InvalidRetryModeError( 

379 provided_retry_mode=retry_mode, 

380 valid_modes="legacy, standard, adaptive", 

381 ) 

382 if retry_mode == "adaptive": 

383 raise InvalidRetryConfigurationError( 

384 retry_config_option="mode=adaptive", 

385 valid_options="mode=legacy, mode=standard", 

386 ) 

387 

388 total_max_attempts = configured_retries.get("total_max_attempts") 

389 if total_max_attempts is not None: 

390 return max(1, int(total_max_attempts)) 

391 

392 max_attempts = configured_retries.get("max_attempts") 

393 if max_attempts is not None: 

394 return max(1, int(max_attempts) + 1) 

395 

396 session_max_attempts = session.get_config_variable("max_attempts") 

397 if session_max_attempts is not None: 397 ↛ 398line 397 didn't jump to line 398 because the condition on line 397 was never true

398 return max(1, int(session_max_attempts)) 

399 

400 if retry_mode == "standard": 

401 return _DEFAULT_STANDARD_MAX_ATTEMPTS 

402 return _DEFAULT_LEGACY_MAX_ATTEMPTS 

403 

404 

405def _retry_delay_seconds(retry_index: int) -> float: 

406 exponential_ceiling = min(2**retry_index, _MAX_RETRY_DELAY_SECONDS) 

407 return random.random() * exponential_ceiling # noqa: S311 

408 

409 

410def _response_is_retryable( 

411 *, 

412 status_code: int, 

413 headers: Mapping[str, Any], 

414 content: bytes, 

415) -> bool: 

416 if status_code in _RETRYABLE_STATUS_CODES: 416 ↛ 418line 416 didn't jump to line 418 because the condition on line 416 was always true

417 return True 

418 if status_code < 400: 

419 return False 

420 

421 normalized_headers = _normalized_headers(headers) 

422 try: 

423 body = _decode_json_body(content) 

424 except (UnicodeDecodeError, ValueError, json.JSONDecodeError): 

425 body = {} 

426 return _error_code(normalized_headers, body) in _RETRYABLE_ERROR_CODES 

427 

428 

429def _checkpoint_params_with_client_token( 

430 params: Mapping[str, Any], 

431) -> dict[str, Any]: 

432 resolved = dict(params) 

433 if resolved.get("ClientToken") is None: 433 ↛ 435line 433 didn't jump to line 435 because the condition on line 433 was always true

434 resolved["ClientToken"] = str(uuid4()) 

435 return resolved 

436 

437 

438class LambdaHttpRequestFactory: 

439 """Build signed Lambda requests without loading a botocore service model.""" 

440 

441 def __init__( 

442 self, 

443 *, 

444 session: Session | None = None, 

445 region_name: str | None = None, 

446 endpoint_url: str | None = None, 

447 user_agent_extra: str | None = None, 

448 use_dualstack_endpoint: bool | None = None, 

449 use_fips_endpoint: bool | None = None, 

450 ignore_configured_endpoint_urls: bool | None = None, 

451 ) -> None: 

452 self.session = session or get_session() 

453 configured_region = _resolve_region(self.session, region_name) 

454 configured_region, legacy_fips_region = _normalize_fips_region( 

455 configured_region 

456 ) 

457 if legacy_fips_region: 

458 use_fips_endpoint = True 

459 endpoint = _resolve_endpoint( 

460 self.session, 

461 region_name=configured_region, 

462 endpoint_url=endpoint_url, 

463 use_dualstack_endpoint=use_dualstack_endpoint, 

464 use_fips_endpoint=use_fips_endpoint, 

465 ignore_configured_endpoint_urls=ignore_configured_endpoint_urls, 

466 ) 

467 self.endpoint_url = endpoint.endpoint_url 

468 self.region_name = endpoint.signing_region 

469 self.signing_name = endpoint.signing_name 

470 self.user_agent = f"durable-execution-sdk-python/{__version__}-async" 

471 if user_agent_extra and self.user_agent not in user_agent_extra: 471 ↛ 472line 471 didn't jump to line 472 because the condition on line 471 was never true

472 self.user_agent = f"{self.user_agent} {user_agent_extra}" 

473 elif user_agent_extra: 

474 self.user_agent = user_agent_extra 

475 

476 def prepare( 

477 self, operation_name: str, params: Mapping[str, Any] 

478 ) -> AWSPreparedRequest: 

479 spec = _request_spec(operation_name, params) 

480 query = { 

481 key: _query_value(value) 

482 for key, value in spec.query.items() 

483 if value is not None 

484 } 

485 query_string = urlencode(query, quote_via=quote, safe="-_.~") 

486 url = f"{self.endpoint_url}{spec.path}" 

487 if query_string: 

488 url = f"{url}?{query_string}" 

489 

490 headers = { 

491 "User-Agent": self.user_agent, 

492 **spec.headers, 

493 } 

494 credentials = self.session.get_credentials() 

495 if credentials is None: 

496 raise NoCredentialsError() 

497 frozen_credentials = credentials.get_frozen_credentials() 

498 

499 request = AWSRequest( 

500 method=spec.method, 

501 url=url, 

502 data=spec.body, 

503 headers=headers, 

504 ) 

505 SigV4Auth( 

506 frozen_credentials, 

507 self.signing_name, 

508 self.region_name, 

509 ).add_auth(request) 

510 return request.prepare() 

511 

512 

513def _normalized_headers(headers: Mapping[str, Any]) -> dict[str, str]: 

514 return {str(key).lower(): str(value) for key, value in headers.items()} 

515 

516 

517def _response_metadata( 

518 status_code: int, 

519 headers: Mapping[str, Any], 

520) -> dict[str, Any]: 

521 normalized = _normalized_headers(headers) 

522 return { 

523 "RequestId": ( 

524 normalized.get("x-amzn-requestid") or normalized.get("x-amzn-request-id") 

525 ), 

526 "HTTPStatusCode": status_code, 

527 "HTTPHeaders": normalized, 

528 "RetryAttempts": 0, 

529 } 

530 

531 

532def _decode_timestamps(value: Any, *, field_name: str | None = None) -> Any: 

533 if ( 

534 field_name is not None 

535 and field_name.endswith("Timestamp") 

536 and isinstance(value, int | float) 

537 ): 

538 return datetime.datetime.fromtimestamp(value, tz=datetime.timezone.utc) 

539 if isinstance(value, list): 

540 return [_decode_timestamps(item) for item in value] 

541 if isinstance(value, dict): 

542 return { 

543 key: _decode_timestamps(item, field_name=key) for key, item in value.items() 

544 } 

545 return value 

546 

547 

548def _decode_json_body(content: bytes) -> dict[str, Any]: 

549 if not content: 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true

550 return {} 

551 decoded = json.loads(content) 

552 if not isinstance(decoded, dict): 552 ↛ 553line 552 didn't jump to line 553 because the condition on line 552 was never true

553 msg = "AWS Lambda returned a non-object JSON response" 

554 raise ValueError(msg) 

555 return cast(dict[str, Any], _decode_timestamps(decoded)) 

556 

557 

558def _error_code(headers: Mapping[str, str], body: Mapping[str, Any]) -> str: 

559 header_code = headers.get("x-amzn-errortype") 

560 if header_code: 560 ↛ 563line 560 didn't jump to line 563 because the condition on line 560 was always true

561 return header_code.split(":", 1)[0].rsplit("#", 1)[-1] 

562 

563 body_code = body.get("__type") or body.get("code") or body.get("Code") 

564 if body_code: 

565 return str(body_code).split(":", 1)[0].rsplit("#", 1)[-1] 

566 return "UnknownError" 

567 

568 

569def parse_lambda_response( 

570 *, 

571 operation_name: str, 

572 status_code: int, 

573 headers: Mapping[str, Any], 

574 content: bytes, 

575 retry_attempts: int = 0, 

576) -> dict[str, Any]: 

577 """Parse a Lambda REST response into the established AWS API mapping.""" 

578 

579 metadata = _response_metadata(status_code, headers) 

580 metadata["RetryAttempts"] = retry_attempts 

581 normalized_headers = cast(dict[str, str], metadata["HTTPHeaders"]) 

582 

583 if status_code < 200 or status_code >= 300: 

584 try: 

585 body = _decode_json_body(content) 

586 except (UnicodeDecodeError, ValueError, json.JSONDecodeError): 

587 body = {} 

588 message = ( 

589 body.get("message") 

590 or body.get("Message") 

591 or content.decode("utf-8", errors="replace") 

592 or f"HTTP {status_code}" 

593 ) 

594 error_response = { 

595 "Error": { 

596 "Code": _error_code(normalized_headers, body), 

597 "Message": str(message), 

598 }, 

599 "ResponseMetadata": metadata, 

600 } 

601 raise ClientError(cast("Any", error_response), operation_name) 

602 

603 if operation_name == "Invoke": 

604 response: dict[str, Any] = { 

605 "StatusCode": status_code, 

606 "Payload": content, 

607 } 

608 invoke_headers = { 

609 "FunctionError": "x-amz-function-error", 

610 "LogResult": "x-amz-log-result", 

611 "ExecutedVersion": "x-amz-executed-version", 

612 "DurableExecutionArn": "x-amz-durable-execution-arn", 

613 } 

614 for output_name, header_name in invoke_headers.items(): 

615 value = normalized_headers.get(header_name) 

616 if value is not None: 

617 response[output_name] = value 

618 else: 

619 response = _decode_json_body(content) 

620 

621 response["ResponseMetadata"] = metadata 

622 return response 

623 

624 

625class BotocoreHttpLambdaClient: 

626 """Synchronous model-free Lambda client using botocore's HTTP session.""" 

627 

628 def __init__( 

629 self, 

630 *, 

631 request_factory: LambdaHttpRequestFactory, 

632 http_session: _SyncHttpSession, 

633 max_attempts: int = 1, 

634 ) -> None: 

635 self._request_factory = request_factory 

636 self._http_session = http_session 

637 self._max_attempts = max_attempts 

638 

639 def _call(self, operation_name: str, **kwargs: Any) -> dict[str, Any]: 

640 for attempt in range(self._max_attempts): 640 ↛ 666line 640 didn't jump to line 666 because the loop on line 640 didn't complete

641 request = self._request_factory.prepare(operation_name, kwargs) 

642 try: 

643 response = self._http_session.send(request) 

644 except Exception: 

645 if attempt + 1 >= self._max_attempts: 

646 raise 

647 time.sleep(_retry_delay_seconds(attempt)) 

648 continue 

649 

650 if attempt + 1 < self._max_attempts and _response_is_retryable( 

651 status_code=response.status_code, 

652 headers=response.headers, 

653 content=response.content, 

654 ): 

655 time.sleep(_retry_delay_seconds(attempt)) 

656 continue 

657 

658 return parse_lambda_response( 

659 operation_name=operation_name, 

660 status_code=response.status_code, 

661 headers=response.headers, 

662 content=response.content, 

663 retry_attempts=attempt, 

664 ) 

665 

666 msg = "Lambda HTTP retry loop exited without a response" 

667 raise RuntimeError(msg) 

668 

669 def checkpoint_durable_execution(self, **kwargs: Any) -> dict[str, Any]: 

670 return self._call( 

671 "CheckpointDurableExecution", 

672 **_checkpoint_params_with_client_token(kwargs), 

673 ) 

674 

675 def get_durable_execution_state(self, **kwargs: Any) -> dict[str, Any]: 

676 return self._call("GetDurableExecutionState", **kwargs) 

677 

678 def invoke(self, **kwargs: Any) -> dict[str, Any]: 

679 return self._call("Invoke", **kwargs) 

680 

681 def get_durable_execution(self, **kwargs: Any) -> dict[str, Any]: 

682 return self._call("GetDurableExecution", **kwargs) 

683 

684 def get_durable_execution_history(self, **kwargs: Any) -> dict[str, Any]: 

685 return self._call("GetDurableExecutionHistory", **kwargs) 

686 

687 def send_durable_execution_callback_success(self, **kwargs: Any) -> dict[str, Any]: 

688 return self._call("SendDurableExecutionCallbackSuccess", **kwargs) 

689 

690 def send_durable_execution_callback_failure(self, **kwargs: Any) -> dict[str, Any]: 

691 return self._call("SendDurableExecutionCallbackFailure", **kwargs) 

692 

693 def send_durable_execution_callback_heartbeat( 

694 self, **kwargs: Any 

695 ) -> dict[str, Any]: 

696 return self._call("SendDurableExecutionCallbackHeartbeat", **kwargs) 

697 

698 def close(self) -> None: 

699 self._http_session.close() 

700 

701 

702class HttpxLambdaClient: 

703 """Asynchronous model-free Lambda client using HTTPX.""" 

704 

705 def __init__( 

706 self, 

707 *, 

708 request_factory: LambdaHttpRequestFactory, 

709 http_client: _AsyncHttpClient, 

710 max_attempts: int = 1, 

711 ) -> None: 

712 self._request_factory = request_factory 

713 self._http_client = http_client 

714 self._max_attempts = max_attempts 

715 

716 async def _call(self, operation_name: str, **kwargs: Any) -> dict[str, Any]: 

717 for attempt in range(self._max_attempts): 717 ↛ 752line 717 didn't jump to line 752 because the loop on line 717 didn't complete

718 request = await asyncio.to_thread( 

719 self._request_factory.prepare, 

720 operation_name, 

721 kwargs, 

722 ) 

723 try: 

724 response = await self._http_client.request( 

725 request.method, 

726 request.url, 

727 content=cast(bytes, request.body or b""), 

728 headers=cast(Mapping[str, str], request.headers), 

729 ) 

730 except Exception: 

731 if attempt + 1 >= self._max_attempts: 731 ↛ 732line 731 didn't jump to line 732 because the condition on line 731 was never true

732 raise 

733 await asyncio.sleep(_retry_delay_seconds(attempt)) 

734 continue 

735 

736 if attempt + 1 < self._max_attempts and _response_is_retryable( 736 ↛ 741line 736 didn't jump to line 741 because the condition on line 736 was never true

737 status_code=response.status_code, 

738 headers=response.headers, 

739 content=response.content, 

740 ): 

741 await asyncio.sleep(_retry_delay_seconds(attempt)) 

742 continue 

743 

744 return parse_lambda_response( 

745 operation_name=operation_name, 

746 status_code=response.status_code, 

747 headers=response.headers, 

748 content=response.content, 

749 retry_attempts=attempt, 

750 ) 

751 

752 msg = "Lambda HTTP retry loop exited without a response" 

753 raise RuntimeError(msg) 

754 

755 async def checkpoint_durable_execution(self, **kwargs: Any) -> dict[str, Any]: 

756 return await self._call( 

757 "CheckpointDurableExecution", 

758 **_checkpoint_params_with_client_token(kwargs), 

759 ) 

760 

761 async def get_durable_execution_state(self, **kwargs: Any) -> dict[str, Any]: 

762 return await self._call("GetDurableExecutionState", **kwargs) 

763 

764 async def invoke(self, **kwargs: Any) -> dict[str, Any]: 

765 return await self._call("Invoke", **kwargs) 

766 

767 async def get_durable_execution(self, **kwargs: Any) -> dict[str, Any]: 

768 return await self._call("GetDurableExecution", **kwargs) 

769 

770 async def get_durable_execution_history(self, **kwargs: Any) -> dict[str, Any]: 

771 return await self._call("GetDurableExecutionHistory", **kwargs) 

772 

773 async def send_durable_execution_callback_success( 

774 self, **kwargs: Any 

775 ) -> dict[str, Any]: 

776 return await self._call("SendDurableExecutionCallbackSuccess", **kwargs) 

777 

778 async def send_durable_execution_callback_failure( 

779 self, **kwargs: Any 

780 ) -> dict[str, Any]: 

781 return await self._call("SendDurableExecutionCallbackFailure", **kwargs) 

782 

783 async def send_durable_execution_callback_heartbeat( 

784 self, **kwargs: Any 

785 ) -> dict[str, Any]: 

786 return await self._call("SendDurableExecutionCallbackHeartbeat", **kwargs) 

787 

788 async def aclose(self) -> None: 

789 await self._http_client.aclose() 

790 

791 

792def create_botocore_http_client( 

793 *, 

794 session: Session | None = None, 

795 region_name: str | None = None, 

796 endpoint_url: str | None = None, 

797 config: Config | None = None, 

798) -> BotocoreHttpLambdaClient: 

799 """Create the default synchronous model-free Lambda HTTP client.""" 

800 

801 resolved_session = session or get_session() 

802 resolved_config = config or Config( 

803 connect_timeout=_DEFAULT_CONNECT_TIMEOUT_SECONDS, 

804 read_timeout=_DEFAULT_READ_TIMEOUT_SECONDS, 

805 ) 

806 config_values = cast("Any", resolved_config) 

807 request_factory = LambdaHttpRequestFactory( 

808 session=resolved_session,