From 675e682a1f1c7606255f40b6840855d31b8bb87e Mon Sep 17 00:00:00 2001 From: Synthes <67906654+Synth3s@users.noreply.github.com> Date: Mon, 8 Dec 2025 17:33:52 +0300 Subject: [PATCH] Fix: Update UserResponseDto for API v2.3.0+ --- remnawave/models/users.py | 403 ++++++++++++++++---------------------- 1 file changed, 165 insertions(+), 238 deletions(-) diff --git a/remnawave/models/users.py b/remnawave/models/users.py index f817d58..a37ea18 100644 --- a/remnawave/models/users.py +++ b/remnawave/models/users.py @@ -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] \ No newline at end of file + 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