BASIS/dynamix-python-sdk
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 14 Wiki Activity

1.3.0

Browse Source
This commit is contained in:
Dmitriy Smirnov
2025-11-25 18:09:46 +03:00
parent 9eac1da63f
commit 1703d1ed33
266 changed files with 5566 additions and 502 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
src/dynamix_sdk/base.py
View File

@@ -28,6 +28,7 @@ from pydantic import (
import yaml
from dynamix_sdk.config import Config, ConfigWithAuth
from dynamix_sdk.exceptions import RequestException
from dynamix_sdk.utils import HTTPMethod, gen_cls_name_from_url_path
@@ -98,9 +99,24 @@ class BaseGetAPIFunctionProtocol(BaseAPIFunctionProtocol):
pass
class BaseDeleteAPIFunctionProtocol(BaseAPIFunctionProtocol):
pass
class BasePatchAPIFunctionProtocol(BaseAPIFunctionProtocol):
pass
class BasePutAPIFunctionProtocol(BaseAPIFunctionProtocol):
pass
base_proto_to_http_method: dict[type[BaseAPIFunctionProtocol], HTTPMethod] = {
BasePostAPIFunctionProtocol: HTTPMethod.POST,
BaseGetAPIFunctionProtocol: HTTPMethod.GET,
BaseDeleteAPIFunctionProtocol: HTTPMethod.DELETE,
BasePatchAPIFunctionProtocol: HTTPMethod.PATCH,
BasePutAPIFunctionProtocol: HTTPMethod.PUT,
}
@@ -328,6 +344,15 @@ def create_api_params_cls(
)
def get_func_api_path(*, func_name: str, path_mapping: dict) -> str:
api_path_parts = []
for func_part in func_name.split('.'):
api_path_parts.append(
path_mapping.get(func_part, func_part)
)
return f'/{"/".join(api_path_parts)}'
APIParamsP = ParamSpec('APIParamsP')
APIResultT = TypeVar(
'APIResultT',
@@ -337,10 +362,9 @@ APIResultT = TypeVar(
class BaseAPI(ABC):
_config: Config
_base_api_path: None | str
_path_mapping_dict: dict[str, str] = path_mapping_dict
_last_call_api_path: None | str = None
_parent_api_group_names: list[str]
_name_mapping_dict: dict[str, str] = name_mapping_dict
_path_mapping_dict: dict[str, str] = path_mapping_dict
_post_json: bool = True
def __init_subclass__(
@@ -359,31 +383,26 @@ class BaseAPI(ABC):
def __init__(
self,
config: Config,
base_api_path: None | str = None
parent_api_group_names: None | list = None,
):
self._config = config
self._base_api_path = base_api_path
self._parent_api_group_names = parent_api_group_names or []
def __getattribute__(self, name: str) -> Any:
if name.startswith('_'):
return super().__getattribute__(name)
else:
path_part = self._path_mapping_dict.get(name, name)
if name in self.__annotations__:
annotation = self.__annotations__[name]
if issubclass(annotation, BaseAPI):
api_cls = annotation
return api_cls(
config=self._config,
base_api_path=(
f'{self._base_api_path or ""}/{path_part}'
parent_api_group_names=(
self._parent_api_group_names + [name]
)
)
else:
self._last_call_api_path = (
f'{self._base_api_path or ""}/{path_part}'
)
attr_value = super().__getattribute__(name)
if not inspect.ismethod(attr_value):
raise ValueError
@@ -406,16 +425,21 @@ class BaseAPI(ABC):
if not isinstance(protocol_method, MethodType):
raise TypeError
func_name_parts = (
self._parent_api_group_names + [protocol_method.__name__]
)
func_name = '.'.join(func_name_parts)
def api_function(
*args: APIParamsP.args,
**kwargs: APIParamsP.kwargs,
):
config = self._config
if self._last_call_api_path:
api_path = self._last_call_api_path
else:
raise ValueError
api_path = get_func_api_path(
func_name=func_name,
path_mapping=self._path_mapping_dict,
)
api_params_cls_name = gen_api_params_cls_name(
api_path=api_path,
@@ -476,7 +500,12 @@ class BaseAPI(ABC):
match http_method:
case HTTPMethod.GET:
req_params = model_data
case HTTPMethod.POST:
case (
HTTPMethod.POST
| HTTPMethod.DELETE
| HTTPMethod.PATCH
| HTTPMethod.PUT
):
if self._post_json:
req_json = model_data
else:
@@ -494,23 +523,33 @@ class BaseAPI(ABC):
).prepare()
attempts = config.http503_attempts
with requests.Session() as session:
while True:
http_response = session.send(
request=http_request,
verify=config.verify_ssl,
)
try:
with requests.Session() as session:
while True:
http_response = session.send(
request=http_request,
verify=config.verify_ssl,
)
if http_response.status_code == 503:
if attempts < 1:
break
if http_response.status_code == 503:
if attempts < 1:
break
else:
attempts -= 1
time.sleep(config.http503_attempts_interval)
else:
attempts -= 1
time.sleep(config.http503_attempts_interval)
else:
break
break
http_response.raise_for_status()
http_response.raise_for_status()
except requests.exceptions.RequestException as e:
if config.wrap_request_exceptions:
raise RequestException(
orig_exception=e,
func_name=func_name,
func_kwargs=kwargs,
)
else:
raise e
api_result_context = APIResultContext(
api_params=api_params,
@@ -553,4 +592,10 @@ class BaseAPI(ABC):
return api_result
api_function.__name__ = func_name.replace('.', '__')
if self._config.f_decorators:
for decorator in reversed(self._config.f_decorators):
api_function = decorator(api_function)
return api_function