mirror of
https://github.com/remnawave/python-sdk.git
synced 2026-08-31 22:31:02 +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.
60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
from typing import Annotated, List
|
|
from uuid import UUID
|
|
|
|
from rapid_api_client import PydanticBody
|
|
|
|
from remnawave.models import (
|
|
BulkDeleteHostsResponseDto,
|
|
BulkDisableHostsResponseDto,
|
|
BulkEnableHostsResponseDto,
|
|
SetInboundToManyHostsRequestDto,
|
|
SetInboundToManyHostsResponseDto,
|
|
SetPortToManyHostsResponseDto,
|
|
)
|
|
from remnawave.rapid import AttributeBody, BaseController, post
|
|
|
|
|
|
class HostsBulkActionsController(BaseController):
|
|
@post("/hosts/bulk/delete", response_class=BulkDeleteHostsResponseDto)
|
|
async def delete_hosts(
|
|
self,
|
|
uuids: Annotated[List[UUID], AttributeBody()],
|
|
) -> BulkDeleteHostsResponseDto:
|
|
"""Delete many hosts"""
|
|
...
|
|
|
|
@post("/hosts/bulk/disable", response_class=BulkDisableHostsResponseDto)
|
|
async def disable_hosts(
|
|
self,
|
|
uuids: Annotated[List[UUID], AttributeBody()],
|
|
) -> BulkDisableHostsResponseDto:
|
|
"""Disable many hosts"""
|
|
...
|
|
|
|
@post("/hosts/bulk/enable", response_class=BulkEnableHostsResponseDto)
|
|
async def enable_hosts(
|
|
self,
|
|
uuids: Annotated[List[UUID], AttributeBody()],
|
|
) -> BulkEnableHostsResponseDto:
|
|
"""Enable many hosts"""
|
|
...
|
|
|
|
@post(
|
|
"/hosts/bulk/set-inbound",
|
|
response_class=SetInboundToManyHostsResponseDto,
|
|
)
|
|
async def set_inbound_to_hosts(
|
|
self,
|
|
body: Annotated[SetInboundToManyHostsRequestDto, PydanticBody()],
|
|
) -> SetInboundToManyHostsResponseDto:
|
|
"""Set inbound to many hosts"""
|
|
...
|
|
|
|
@post("/hosts/bulk/set-port", response_class=SetPortToManyHostsResponseDto)
|
|
async def set_port_to_hosts(
|
|
self,
|
|
uuids: Annotated[List[UUID], AttributeBody()],
|
|
port: Annotated[float, AttributeBody()],
|
|
) -> SetPortToManyHostsResponseDto:
|
|
"""Set port to many hosts"""
|
|
...
|