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