Remnawave_python-sdk/remnawave/controllers/inbounds_bulk_actions.py
Artem 950a7ffab3
Add Pydantic models for nodes usage history, subscriptions, and user management
- 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.
2025-07-03 12:22:16 +02:00

50 lines
1.5 KiB
Python

from typing import Annotated
from uuid import UUID
from remnawave.models import (
AddInboundToNodesResponseDto,
AddInboundToUsersResponseDto,
RemoveInboundFromNodesResponseDto,
RemoveInboundFromUsersResponseDto,
)
from remnawave.rapid import AttributeBody, BaseController, post
class InboundsBulkActionsController(BaseController):
@post("/inbounds/bulk/add-to-users", response_class=AddInboundToUsersResponseDto)
async def add_inbound_to_users(
self,
inbound_uuid: Annotated[UUID, AttributeBody()],
) -> AddInboundToUsersResponseDto:
"""Add Inbound To Users"""
...
@post(
"/inbounds/bulk/remove-from-users",
response_class=RemoveInboundFromUsersResponseDto,
)
async def remove_inbound_from_users(
self,
inbound_uuid: Annotated[UUID, AttributeBody()],
) -> RemoveInboundFromUsersResponseDto:
"""Remove Inbound From Users"""
...
@post("/inbounds/bulk/add-to-nodes", response_class=AddInboundToNodesResponseDto)
async def add_inbound_to_nodes(
self,
inbound_uuid: Annotated[UUID, AttributeBody()],
) -> AddInboundToNodesResponseDto:
"""Add Inbound To All Nodes"""
...
@post(
"/inbounds/bulk/remove-from-nodes",
response_class=RemoveInboundFromNodesResponseDto,
)
async def remove_inbound_from_nodes(
self,
inbound_uuid: Annotated[UUID, AttributeBody()],
) -> RemoveInboundFromNodesResponseDto:
"""Remove Inbound From All Nodes"""
...