mirror of
https://github.com/remnawave/python-sdk.git
synced 2026-10-20 19:50:44 +00:00
Fix: Update UserResponseDto for API v2.3.0+
This commit is contained in:
Synthes
GitHub
parent
4db922bf74
commit
675e682a1f
1 changed files with 165 additions and 238 deletions
|
|
@ -4,14 +4,14 @@ from uuid import UUID
|
|||
|
||||
from pydantic import (
|
||||
BaseModel,
|
||||
EmailStr,
|
||||
Field,
|
||||
RootModel,
|
||||
StringConstraints,
|
||||
model_validator,
|
||||
)
|
||||
from pydantic.alias_generators import to_camel
|
||||
|
||||
from remnawave.enums import TrafficLimitStrategy, UserStatus
|
||||
from remnawave.utils.happ_crypt import create_happ_crypto_link
|
||||
|
||||
|
||||
class UserActiveInboundsDto(BaseModel):
|
||||
|
|
@ -32,273 +32,121 @@ class ActiveInternalSquadDto(BaseModel):
|
|||
name: str
|
||||
|
||||
|
||||
class HappCrypto(BaseModel):
|
||||
crypto_link: str = Field(alias="cryptoLink")
|
||||
class UserTrafficDto(BaseModel):
|
||||
"""Traffic statistics nested in userTraffic object (API v2.3.0+)"""
|
||||
used_traffic_bytes: float = Field(alias="usedTrafficBytes")
|
||||
lifetime_used_traffic_bytes: float = Field(alias="lifetimeUsedTrafficBytes")
|
||||
online_at: datetime | None = Field(None, alias="onlineAt")
|
||||
first_connected_at: datetime | None = Field(None, alias="firstConnectedAt")
|
||||
last_connected_node_uuid: UUID | None = Field(None, alias="lastConnectedNodeUuid")
|
||||
|
||||
model_config = {"alias_generator": to_camel, "populate_by_name": True}
|
||||
|
||||
|
||||
class CreateUserRequestDto(BaseModel):
|
||||
"""Request DTO for creating a user"""
|
||||
# Required fields
|
||||
expire_at: datetime = Field(..., serialization_alias="expireAt")
|
||||
username: Annotated[
|
||||
str,
|
||||
StringConstraints(pattern=r"^[a-zA-Z0-9_-]+$", min_length=3, max_length=36)
|
||||
] = Field(..., description="Unique username for the user")
|
||||
expire_at: datetime = Field(..., serialization_alias="expireAt", description="Account expiration date")
|
||||
|
||||
# Optional fields with defaults
|
||||
status: UserStatus = Field(default=UserStatus.ACTIVE, description="User account status")
|
||||
traffic_limit_strategy: TrafficLimitStrategy = Field(
|
||||
default=TrafficLimitStrategy.NO_RESET,
|
||||
serialization_alias="trafficLimitStrategy",
|
||||
description="Traffic reset strategy"
|
||||
str, StringConstraints(pattern=r"^[a-zA-Z0-9_-]+$", min_length=3, max_length=36)
|
||||
]
|
||||
created_at: datetime | None = Field(None, serialization_alias="createdAt")
|
||||
status: UserStatus | None = None
|
||||
short_uuid: str | None = Field(None, serialization_alias="shortUuid")
|
||||
trojan_password: Annotated[
|
||||
str | None, StringConstraints(min_length=8, max_length=32)
|
||||
] = Field(None, serialization_alias="trojanPassword")
|
||||
vless_uuid: str | None = Field(None, serialization_alias="vlessUuid")
|
||||
ss_password: Annotated[
|
||||
str | None, StringConstraints(min_length=8, max_length=32)
|
||||
] = Field(None, serialization_alias="ssPassword")
|
||||
traffic_limit_bytes: int | None = Field(
|
||||
None, serialization_alias="trafficLimitBytes", strict=True, ge=0
|
||||
)
|
||||
|
||||
# Optional fields
|
||||
short_uuid: Optional[str] = Field(None, serialization_alias="shortUuid")
|
||||
trojan_password: Optional[Annotated[str, StringConstraints(min_length=8, max_length=32)]] = Field(
|
||||
None, serialization_alias="trojanPassword"
|
||||
traffic_limit_strategy: TrafficLimitStrategy | None = Field(
|
||||
None, serialization_alias="trafficLimitStrategy"
|
||||
)
|
||||
vless_uuid: Optional[UUID] = Field(None, serialization_alias="vlessUuid")
|
||||
ss_password: Optional[Annotated[str, StringConstraints(min_length=8, max_length=32)]] = Field(
|
||||
None, serialization_alias="ssPassword"
|
||||
last_traffic_reset_at: datetime | None = Field(
|
||||
None, serialization_alias="lastTrafficResetAt"
|
||||
)
|
||||
traffic_limit_bytes: Optional[int] = Field(
|
||||
None, serialization_alias="trafficLimitBytes", ge=0
|
||||
description: str | None = None
|
||||
tag: str | None = None
|
||||
telegram_id: int | None = Field(None, serialization_alias="telegramId")
|
||||
email: str | None = None
|
||||
hwid_device_limit: int | None = Field(
|
||||
None, serialization_alias="hwidDeviceLimit", strict=True, ge=0
|
||||
)
|
||||
created_at: Optional[datetime] = Field(None, serialization_alias="createdAt")
|
||||
last_traffic_reset_at: Optional[datetime] = Field(None, serialization_alias="lastTrafficResetAt")
|
||||
description: Optional[str] = None
|
||||
tag: Optional[Annotated[str, StringConstraints(max_length=16, pattern=r"^[A-Z0-9_]+$")]] = None
|
||||
telegram_id: Optional[int] = Field(None, serialization_alias="telegramId")
|
||||
email: Optional[EmailStr] = None
|
||||
hwid_device_limit: Optional[int] = Field(None, serialization_alias="hwidDeviceLimit", ge=0)
|
||||
active_internal_squads: Optional[List[UUID]] = Field(None, serialization_alias="activeInternalSquads")
|
||||
uuid: Optional[UUID] = Field(
|
||||
None,
|
||||
description="Optional. Pass UUID to create user with specific UUID, otherwise it will be generated automatically."
|
||||
active_internal_squads: list[UUID] | None = Field(
|
||||
None, serialization_alias="activeInternalSquads"
|
||||
)
|
||||
external_squad_uuid: Optional[UUID] = Field(None, serialization_alias="externalSquadUuid")
|
||||
external_squad_uuid: UUID | None = Field(None, serialization_alias="externalSquadUuid")
|
||||
uuid: Optional[UUID] = Field(None, description="UUID of the user. Optional. If not provided, a new UUID will be generated by Remnawave.")
|
||||
|
||||
|
||||
class UpdateUserRequestDto(BaseModel):
|
||||
"""Request DTO for updating a user"""
|
||||
# Either username or uuid must be provided, uuid has priority
|
||||
username: Optional[str] = Field(None, description="Username of the user")
|
||||
uuid: Optional[UUID] = Field(
|
||||
None,
|
||||
description="UUID of the user. UUID has higher priority than username"
|
||||
uuid: UUID
|
||||
description: str | None = None
|
||||
email: str | None = None
|
||||
expire_at: datetime | None = Field(None, serialization_alias="expireAt")
|
||||
hwid_device_limit: int | None = Field(
|
||||
None, serialization_alias="hwidDeviceLimit", strict=True, ge=0
|
||||
)
|
||||
|
||||
# Optional update fields
|
||||
status: Optional[UserStatus] = None
|
||||
description: Optional[str] = None
|
||||
email: Optional[EmailStr] = None
|
||||
expire_at: Optional[datetime] = Field(None, serialization_alias="expireAt")
|
||||
hwid_device_limit: Optional[int] = Field(None, serialization_alias="hwidDeviceLimit", ge=0)
|
||||
tag: Optional[Annotated[str, StringConstraints(max_length=16, pattern=r"^[A-Z0-9_]+$")]] = None
|
||||
telegram_id: Optional[int] = Field(None, serialization_alias="telegramId")
|
||||
traffic_limit_bytes: Optional[int] = Field(None, serialization_alias="trafficLimitBytes", ge=0)
|
||||
traffic_limit_strategy: Optional[TrafficLimitStrategy] = Field(
|
||||
status: UserStatus | None = None
|
||||
tag: str | None = None
|
||||
telegram_id: int | None = Field(None, serialization_alias="telegramId")
|
||||
traffic_limit_bytes: int | None = Field(
|
||||
None, serialization_alias="trafficLimitBytes", strict=True, ge=0
|
||||
)
|
||||
traffic_limit_strategy: TrafficLimitStrategy | None = Field(
|
||||
None, serialization_alias="trafficLimitStrategy"
|
||||
)
|
||||
active_internal_squads: Optional[List[UUID]] = Field(
|
||||
active_internal_squads: list[UUID] | None = Field(
|
||||
None, serialization_alias="activeInternalSquads"
|
||||
)
|
||||
external_squad_uuid: Optional[UUID] = Field(None, serialization_alias="externalSquadUuid")
|
||||
|
||||
|
||||
class UserTrafficDto(BaseModel):
|
||||
"""User traffic information"""
|
||||
used_traffic_bytes: float = Field(alias="usedTrafficBytes")
|
||||
lifetime_used_traffic_bytes: float = Field(alias="lifetimeUsedTrafficBytes")
|
||||
online_at: Optional[datetime] = Field(None, alias="onlineAt")
|
||||
first_connected_at: Optional[datetime] = Field(None, alias="firstConnectedAt")
|
||||
last_connected_node_uuid: Optional[UUID] = Field(None, alias="lastConnectedNodeUuid")
|
||||
external_squad_uuid: UUID | None = Field(None, alias="externalSquadUuid")
|
||||
|
||||
|
||||
class UserResponseDto(BaseModel):
|
||||
"""User response DTO - обновленная структура с userTraffic"""
|
||||
"""User response model for API v2.3.0+"""
|
||||
uuid: UUID
|
||||
id: int
|
||||
short_uuid: str = Field(alias="shortUuid")
|
||||
username: str
|
||||
status: UserStatus = Field(default=UserStatus.ACTIVE)
|
||||
traffic_limit_bytes: int = Field(0, alias="trafficLimitBytes")
|
||||
traffic_limit_strategy: TrafficLimitStrategy = Field(
|
||||
TrafficLimitStrategy.NO_RESET, alias="trafficLimitStrategy"
|
||||
)
|
||||
status: UserStatus | None = None
|
||||
traffic_limit_bytes: int | None = Field(None, alias="trafficLimitBytes", default=0)
|
||||
traffic_limit_strategy: str | None = Field(None, alias="trafficLimitStrategy", default="NO_RESET")
|
||||
sub_last_user_agent: str | None = Field(None, alias="subLastUserAgent")
|
||||
sub_last_opened_at: datetime | None = Field(None, alias="subLastOpenedAt")
|
||||
expire_at: datetime = Field(alias="expireAt")
|
||||
telegram_id: Optional[int] = Field(None, alias="telegramId")
|
||||
email: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
tag: Optional[str] = None
|
||||
hwid_device_limit: Optional[int] = Field(None, alias="hwidDeviceLimit")
|
||||
external_squad_uuid: Optional[UUID] = Field(None, alias="externalSquadUuid")
|
||||
sub_revoked_at: datetime | None = Field(None, alias="subRevokedAt")
|
||||
last_traffic_reset_at: datetime | None = Field(None, alias="lastTrafficResetAt")
|
||||
trojan_password: str = Field(alias="trojanPassword")
|
||||
vless_uuid: UUID = Field(alias="vlessUuid")
|
||||
ss_password: str = Field(alias="ssPassword")
|
||||
last_trigger_threshold: int = Field(0, alias="lastTriggeredThreshold")
|
||||
sub_revoked_at: Optional[datetime] = Field(None, alias="subRevokedAt")
|
||||
sub_last_user_agent: Optional[str] = Field(None, alias="subLastUserAgent")
|
||||
sub_last_opened_at: Optional[datetime] = Field(None, alias="subLastOpenedAt")
|
||||
last_traffic_reset_at: Optional[datetime] = Field(None, alias="lastTrafficResetAt")
|
||||
description: str | None = None
|
||||
telegram_id: int | None = Field(None, alias="telegramId")
|
||||
email: str | None = None
|
||||
hwid_device_limit: int | None = Field(
|
||||
None, alias="hwidDeviceLimit", strict=True, ge=0
|
||||
)
|
||||
active_internal_squads: list[ActiveInternalSquadDto] = Field(alias="activeInternalSquads", default_factory=list)
|
||||
subscription_url: str = Field(alias="subscriptionUrl")
|
||||
last_trigger_threshold: int | None = Field(None, alias="lastTriggeredThreshold", default=0)
|
||||
tag: str | None = Field(None, alias="tag")
|
||||
external_squad_uuid: UUID | None = Field(None, alias="externalSquadUuid")
|
||||
created_at: datetime = Field(alias="createdAt")
|
||||
updated_at: datetime = Field(alias="updatedAt")
|
||||
subscription_url: str = Field(alias="subscriptionUrl")
|
||||
active_internal_squads: list[ActiveInternalSquadDto] = Field(alias="activeInternalSquads")
|
||||
user_traffic: UserTrafficDto = Field(alias="userTraffic")
|
||||
|
||||
@property
|
||||
def used_traffic_bytes(self) -> float:
|
||||
"""Backward compatibility property"""
|
||||
return self.user_traffic.used_traffic_bytes
|
||||
@model_validator(mode='before')
|
||||
@classmethod
|
||||
def handle_response_wrapper(cls, data: dict) -> dict:
|
||||
"""Handle response wrapper (API v2.3.0+)"""
|
||||
if isinstance(data, dict) and "response" in data:
|
||||
return data["response"]
|
||||
return data
|
||||
|
||||
@property
|
||||
def lifetime_used_traffic_bytes(self) -> float:
|
||||
"""Backward compatibility property"""
|
||||
return self.user_traffic.lifetime_used_traffic_bytes
|
||||
|
||||
@property
|
||||
def online_at(self) -> Optional[datetime]:
|
||||
"""Backward compatibility property"""
|
||||
return self.user_traffic.online_at
|
||||
|
||||
@property
|
||||
def first_connected(self) -> Optional[datetime]:
|
||||
"""Backward compatibility property"""
|
||||
return self.user_traffic.first_connected_at
|
||||
|
||||
@property
|
||||
def last_connected_node_uuid(self) -> Optional[UUID]:
|
||||
"""Backward compatibility property"""
|
||||
return self.user_traffic.last_connected_node_uuid
|
||||
|
||||
@property
|
||||
def happ(self) -> HappCrypto:
|
||||
"""Generate Happ Crypto Link"""
|
||||
crypto_link = create_happ_crypto_link(self.subscription_url)
|
||||
return HappCrypto(cryptoLink=crypto_link)
|
||||
|
||||
|
||||
class RevokeUserRequestDto(BaseModel):
|
||||
"""Request DTO for revoking user subscription"""
|
||||
short_uuid: Optional[str] = Field(
|
||||
None,
|
||||
serialization_alias="shortUuid",
|
||||
description="Optional. If not provided, a new short UUID will be generated by Remnawave.",
|
||||
min_length=6,
|
||||
max_length=48,
|
||||
pattern=r"^[a-zA-Z0-9_-]+$",
|
||||
)
|
||||
|
||||
|
||||
class SubscriptionRequestRecord(BaseModel):
|
||||
"""Subscription request history record"""
|
||||
id: int
|
||||
user_uuid: UUID = Field(alias="userUuid")
|
||||
request_at: datetime = Field(alias="requestAt")
|
||||
request_ip: Optional[str] = Field(alias="requestIp")
|
||||
user_agent: Optional[str] = Field(alias="userAgent")
|
||||
|
||||
|
||||
class SubscriptionRequestsResponseData(BaseModel):
|
||||
"""Subscription requests response data"""
|
||||
total: int
|
||||
records: List[SubscriptionRequestRecord]
|
||||
|
||||
|
||||
class CreateUserResponseDto(UserResponseDto):
|
||||
"""Response for create user"""
|
||||
pass
|
||||
|
||||
|
||||
class UpdateUserResponseDto(UserResponseDto):
|
||||
"""Response for update user"""
|
||||
pass
|
||||
|
||||
|
||||
class GetUserByUuidResponseDto(UserResponseDto):
|
||||
"""Response for get user by UUID"""
|
||||
pass
|
||||
|
||||
|
||||
class GetUserByShortUuidResponseDto(UserResponseDto):
|
||||
"""Response for get user by short UUID"""
|
||||
pass
|
||||
|
||||
|
||||
class GetUserByUsernameResponseDto(UserResponseDto):
|
||||
"""Response for get user by username"""
|
||||
pass
|
||||
|
||||
|
||||
class GetUserByIdResponseDto(UserResponseDto):
|
||||
"""Response for get user by ID"""
|
||||
pass
|
||||
|
||||
|
||||
class DisableUserResponseDto(UserResponseDto):
|
||||
"""Response for disable user"""
|
||||
pass
|
||||
|
||||
|
||||
class EnableUserResponseDto(UserResponseDto):
|
||||
"""Response for enable user"""
|
||||
pass
|
||||
|
||||
|
||||
class ResetUserTrafficResponseDto(UserResponseDto):
|
||||
"""Response for reset user traffic"""
|
||||
pass
|
||||
|
||||
|
||||
class RevokeUserSubscriptionResponseDto(UserResponseDto):
|
||||
"""Response for revoke user subscription"""
|
||||
pass
|
||||
|
||||
|
||||
class ActivateAllInboundsResponseDto(UserResponseDto):
|
||||
"""Response for activate all inbounds"""
|
||||
pass
|
||||
|
||||
|
||||
class UsersResponseDto(BaseModel):
|
||||
"""Users collection response"""
|
||||
users: list[UserResponseDto]
|
||||
total: int
|
||||
|
||||
|
||||
class GetAllUsersResponseDto(UsersResponseDto):
|
||||
"""Response for get all users"""
|
||||
pass
|
||||
|
||||
|
||||
class TagsResponseDto(BaseModel):
|
||||
"""Tags collection response"""
|
||||
tags: list[str]
|
||||
|
||||
|
||||
class GetAllTagsResponseDto(TagsResponseDto):
|
||||
"""Response for get all tags"""
|
||||
pass
|
||||
|
||||
|
||||
class GetSubscriptionRequestsResponseDto(SubscriptionRequestsResponseData):
|
||||
"""Response for get subscription requests"""
|
||||
pass
|
||||
|
||||
|
||||
class GetUserSubscriptionRequestHistoryResponseDto(SubscriptionRequestsResponseData):
|
||||
"""Response for get user subscription request history"""
|
||||
pass
|
||||
|
||||
|
||||
class DeleteUserResponseDto(BaseModel):
|
||||
"""Response for delete user"""
|
||||
is_deleted: bool = Field(alias="isDeleted")
|
||||
|
||||
model_config = {"alias_generator": to_camel, "populate_by_name": True}
|
||||
|
||||
class EmailUserResponseDto(RootModel[list[UserResponseDto]]):
|
||||
"""Response for get users by email"""
|
||||
def __iter__(self):
|
||||
return iter(self.root)
|
||||
|
||||
|
|
@ -307,7 +155,6 @@ class EmailUserResponseDto(RootModel[list[UserResponseDto]]):
|
|||
|
||||
|
||||
class TagUserResponseDto(RootModel[list[UserResponseDto]]):
|
||||
"""Response for get users by tag"""
|
||||
def __iter__(self):
|
||||
return iter(self.root)
|
||||
|
||||
|
|
@ -316,9 +163,89 @@ class TagUserResponseDto(RootModel[list[UserResponseDto]]):
|
|||
|
||||
|
||||
class TelegramUserResponseDto(RootModel[list[UserResponseDto]]):
|
||||
"""Response for get users by telegram ID"""
|
||||
def __iter__(self):
|
||||
return iter(self.root)
|
||||
|
||||
def __getitem__(self, item):
|
||||
return self.root[item]
|
||||
return self.root[item]
|
||||
|
||||
|
||||
class UsersResponseDto(BaseModel):
|
||||
users: list[UserResponseDto]
|
||||
total: int
|
||||
|
||||
|
||||
class DeleteUserResponseDto(BaseModel):
|
||||
is_deleted: bool = Field(alias="isDeleted")
|
||||
|
||||
|
||||
class TagsResponseDto(BaseModel):
|
||||
tags: list[str]
|
||||
|
||||
|
||||
class CreateUserResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class UpdateUserResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class DisableUserResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class EnableUserResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class ResetUserTrafficResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class RevokeUserSubscriptionResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class ActivateAllInboundsResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class GetUserByUuidResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class GetUserByShortUuidResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class GetUserByUsernameResponseDto(UserResponseDto):
|
||||
pass
|
||||
|
||||
|
||||
class RevokeUserRequestDto(BaseModel):
|
||||
short_uuid: str | None = Field(
|
||||
None,
|
||||
serialization_alias="shortUuid",
|
||||
description="Optional. If not provided, a new short UUID will be generated by Remnawave. Please note that it is strongly recommended to allow Remnawave to generate the short UUID.",
|
||||
min_length=6,
|
||||
max_length=48,
|
||||
pattern=r"^[a-zA-Z0-9_-]+$",
|
||||
)
|
||||
|
||||
|
||||
class SubscriptionRequestRecord(BaseModel):
|
||||
id: int
|
||||
user_uuid: UUID = Field(alias="userUuid")
|
||||
request_at: datetime = Field(alias="requestAt")
|
||||
request_ip: Optional[str] = Field(alias="requestIp")
|
||||
user_agent: Optional[str] = Field(alias="userAgent")
|
||||
|
||||
|
||||
class SubscriptionRequestsResponseData(BaseModel):
|
||||
total: int
|
||||
records: List[SubscriptionRequestRecord]
|
||||
|
||||
|
||||
class GetSubscriptionRequestsResponseDto(SubscriptionRequestsResponseData):
|
||||
pass
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue