mirror of
https://github.com/remnawave/python-sdk.git
synced 2026-08-04 14:37:36 +00:00
950a7ffab3
- Implemented models for nodes usage history including NodeInfoDto, GetUserAccessibleNodesResponse, and usage statistics. - Created subscription-related models such as UserSubscription, SubscriptionInfoData, and various response DTOs for subscription information. - Developed models for subscription settings, templates, and system statistics, enhancing the API's capability to manage and retrieve subscription configurations. - Introduced user management models including CreateUserRequestDto, UpdateUserRequestDto, and various response DTOs for user actions. - Added bulk actions for user updates and statistics tracking, improving the efficiency of user management operations. - Established a base controller for handling API requests and responses, along with decorators for HTTP methods to streamline API interactions. - Included utility functions for serialization using orjson for better performance with Pydantic models.
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from functools import partial, wraps
|
|
from inspect import signature
|
|
from typing import Any, Callable, Coroutine, Type
|
|
|
|
from httpx import AsyncClient, Response
|
|
from pydantic import TypeAdapter
|
|
from rapid_api_client.typing import BM, T
|
|
|
|
from .client import BaseController, CustomRapidParameters
|
|
|
|
|
|
def http(
|
|
method: str,
|
|
path: str,
|
|
response_class: Type[BM | str | bytes | Response] | TypeAdapter[T] = Response,
|
|
timeout: float | None = None,
|
|
) -> Callable[
|
|
[Callable], Callable[..., Coroutine[Any, Any, BM | str | bytes | Response | T]]
|
|
]:
|
|
def decorator(
|
|
func: Callable,
|
|
) -> Callable[..., Coroutine[Any, Any, BM | str | bytes | Response | T]]:
|
|
sig = signature(func)
|
|
rapid_parameters = CustomRapidParameters.from_sig(sig)
|
|
|
|
@wraps(func)
|
|
async def wrapper(
|
|
api: BaseController, *args, **kwargs
|
|
) -> BM | str | bytes | Response | T:
|
|
assert isinstance(
|
|
api, BaseController
|
|
), f"{api} should be an instance of BaseController"
|
|
assert isinstance(
|
|
api.client, AsyncClient
|
|
), f"{api.client} should be an instance of httpx.AsyncClient"
|
|
|
|
# noinspection PyProtectedMember
|
|
request = api._build_request(
|
|
sig, rapid_parameters, method, path, (api,) + args, kwargs, timeout
|
|
)
|
|
response = await api.client.send(request)
|
|
# noinspection PyProtectedMember
|
|
return api._handle_response(response, response_class=response_class)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
get = partial(http, "GET")
|
|
post = partial(http, "POST")
|
|
delete = partial(http, "DELETE")
|
|
put = partial(http, "PUT")
|
|
patch = partial(http, "PATCH")
|