1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
|
"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT."""
import requests
from .types import SDKInitHook, BeforeRequestContext, BeforeRequestHook, AfterSuccessContext, AfterSuccessHook, AfterErrorContext, AfterErrorHook, Hooks
from .registration import init_hooks
from typing import List, Optional, Tuple
class SDKHooks(Hooks):
def __init__(self):
self.sdk_init_hooks: List[SDKInitHook] = []
self.before_request_hooks: List[BeforeRequestHook] = []
self.after_success_hooks: List[AfterSuccessHook] = []
self.after_error_hooks: List[AfterErrorHook] = []
init_hooks(self)
def register_sdk_init_hook(self, hook: SDKInitHook) -> None:
self.sdk_init_hooks.append(hook)
def register_before_request_hook(self, hook: BeforeRequestHook) -> None:
self.before_request_hooks.append(hook)
def register_after_success_hook(self, hook: AfterSuccessHook) -> None:
self.after_success_hooks.append(hook)
def register_after_error_hook(self, hook: AfterErrorHook) -> None:
self.after_error_hooks.append(hook)
def sdk_init(self, base_url: str, client: requests.Session) -> Tuple[str, requests.Session]:
for hook in self.sdk_init_hooks:
base_url, client = hook.sdk_init(base_url, client)
return base_url, client
def before_request(self, hook_ctx: BeforeRequestContext, request: requests.PreparedRequest) -> requests.PreparedRequest:
for hook in self.before_request_hooks:
out = hook.before_request(hook_ctx, request)
if isinstance(out, Exception):
raise out
request = out
return request
def after_success(self, hook_ctx: AfterSuccessContext, response: requests.Response) -> requests.Response:
for hook in self.after_success_hooks:
out = hook.after_success(hook_ctx, response)
if isinstance(out, Exception):
raise out
response = out
return response
def after_error(self, hook_ctx: AfterErrorContext, response: Optional[requests.Response], error: Optional[Exception]) -> Tuple[Optional[requests.Response], Optional[Exception]]:
for hook in self.after_error_hooks:
result = hook.after_error(hook_ctx, response, error)
if isinstance(result, Exception):
raise result
response, error = result
return response, error
|
