Remzona54/Remnawave_python-sdk
1
0
Fork 0
mirror of https://github.com/remnawave/python-sdk.git synced 2026-05-13 12:16:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #10 from remnawave/development

Browse source 
Bump version, fix models and add new route
This commit is contained in:
Artem 2025-09-16 21:51:42 +02:00 • committed by GitHub
commit 600628baf6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 182 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

3
README.md
View file

@ -63,7 +63,8 @@ pip install git+https://github.com/remnawave/python-sdk.git@development
| Contract Version | Remnawave Panel Version |
| ---------------- | ----------------------- |
| 2.1.9 | >=2.1.9 |
| 2.1.13 | >=2.1.13 |
| 2.1.9 | >=2.1.9, <=2.1.12 |
| 2.1.8 | ==2.1.8 |
| 2.1.7.post1 | ==2.1.7 |
| 2.1.4 | >=2.1.4, <2.1.7 |

6
pyproject.toml
View file

@ -1,7 +1,7 @@
[project]
name = "remnawave"
version = "2.1.9"
description = "A Python SDK for interacting with the Remnawave API v2.1.9."
version = "2.1.13"
description = "A Python SDK for interacting with the Remnawave API v2.1.13."
authors = [
{name = "Artem",email = "dev@forestsnet.com"}
]
@ -56,4 +56,4 @@ asyncio_default_fixture_loop_scope = "function"
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"
build-backend = "poetry.core.masonry.api"

34
remnawave/controllers/hwid.py
View file

@ -5,14 +5,32 @@ from remnawave.models import (
CreateUserHwidDeviceResponseDto,
DeleteUserHwidDeviceResponseDto,
GetUserHwidDevicesResponseDto,
GetHwidStatisticsResponseDto,
CreateHWIDUser,
HWIDDeleteRequest
HWIDDeleteRequest,
DeleteUserAllHwidDeviceRequestDto
)
from rapid_api_client import Path, PydanticBody
from remnawave.rapid import AttributeBody, BaseController, post, get
class HWIDUserController(BaseController):
@get("/hwid/devices", response_class=GetUserHwidDevicesResponseDto)
async def get_hwid_users(
self,
size: Annotated[int | None, AttributeBody()] = None,
start: Annotated[int | None, AttributeBody()] = None,
) -> GetUserHwidDevicesResponseDto:
"""Get all user HWID devices"""
...
@get("/hwid/devices/stats", response_class=GetHwidStatisticsResponseDto)
async def get_hwid_stats(
self,
) -> GetHwidStatisticsResponseDto:
"""Get HWID statistics"""
...
@post("/hwid/devices", response_class=CreateUserHwidDeviceResponseDto)
async def add_hwid_to_users(
self,
@ -20,7 +38,7 @@ class HWIDUserController(BaseController):
) -> CreateUserHwidDeviceResponseDto:
"""Create a user HWID device"""
...
@post("/hwid/devices/delete", response_class=DeleteUserHwidDeviceResponseDto)
async def delete_hwid_to_user(
self,
@ -28,11 +46,19 @@ class HWIDUserController(BaseController):
) -> DeleteUserHwidDeviceResponseDto:
"""Delete a user HWID device"""
...
@post("/hwid/devices/delete-all", response_class=DeleteUserHwidDeviceResponseDto)
async def delete_all_hwid_user(
self,
body: Annotated[DeleteUserAllHwidDeviceRequestDto, PydanticBody()],
) -> DeleteUserHwidDeviceResponseDto:
"""Delete all user HWID devices"""
...
@get("/hwid/devices/{uuid}", response_class=GetUserHwidDevicesResponseDto)
async def get_hwid_user(
self,
uuid: Annotated[str, Path(description="UUID of the User")],
) -> GetUserHwidDevicesResponseDto:
"""Get a user HWID device"""
...
...

4
remnawave/models/__init__.py
View file

@ -74,6 +74,8 @@ from .hwid import (
HWIDDeleteRequest, # Legacy alias
HWIDUserResponseDto, # Legacy alias
HWIDUserResponseDtoList, # Legacy alias
GetHwidStatisticsResponseDto,
DeleteUserAllHwidDeviceRequestDto
)
from .inbounds import (
AllInboundsData,
@ -346,6 +348,8 @@ __all__ = [
"HWIDDeleteRequest", # Legacy alias
"HWIDUserResponseDto", # Legacy alias
"HWIDUserResponseDtoList", # Legacy alias
"GetHwidStatisticsResponseDto",
"DeleteUserAllHwidDeviceRequestDto",
# Bandwidth stats models
"GetNodeUserUsageByRangeResponseDto",
"GetNodesRealtimeUsageResponseDto",

36
remnawave/models/hwid.py
View file

@ -31,28 +31,54 @@ class HwidDeviceDto(BaseModel):
class HwidDevicesData(BaseModel):
total: float
total: int
devices: List[HwidDeviceDto]
class CreateUserHwidDeviceResponseDto(BaseModel):
total: float
total: int
devices: List[HwidDeviceDto]
class DeleteUserHwidDeviceResponseDto(BaseModel):
total: float
total: int
devices: List[HwidDeviceDto]
class GetUserHwidDevicesResponseDto(BaseModel):
total: float
total: int
devices: List[HwidDeviceDto]
class PlatformStatItem(BaseModel):
platform: str
count: float
class AppStatItem(BaseModel):
app: str
count: float
class HwidStats(BaseModel):
total_unique_devices: float = Field(alias="totalUniqueDevices")
total_hwid_devices: float = Field(alias="totalHwidDevices")
average_hwid_devices_per_user: float = Field(alias="averageHwidDevicesPerUser")
class HwidStatisticsData(BaseModel):
by_platform: List[PlatformStatItem] = Field(alias="byPlatform")
by_app: List[AppStatItem] = Field(alias="byApp")
stats: HwidStats
class GetHwidStatisticsResponseDto(HwidStatisticsData):
pass
class DeleteUserAllHwidDeviceRequestDto(BaseModel):
user_uuid: UUID = Field(serialization_alias="userUuid")
# Legacy aliases for backward compatibility
CreateHWIDUser = CreateUserHwidDeviceRequestDto
HWIDUserResponseDto = HwidDeviceDto
HWIDUserResponseDtoList = HwidDevicesData
HWIDDeleteRequest = DeleteUserHwidDeviceRequestDto

112
remnawave/models/users.py
View file

@ -1,5 +1,5 @@
from datetime import datetime
from typing import Annotated, List, Optional
from typing import Annotated
from uuid import UUID
from pydantic import (
@ -16,8 +16,8 @@ class UserActiveInboundsDto(BaseModel):
uuid: UUID
tag: str
type: str
network: Optional[str] = None
security: Optional[str] = None
network: str | None = None
security: str | None = None
class UserLastConnectedNodeDto(BaseModel):
@ -39,103 +39,99 @@ class CreateUserRequestDto(BaseModel):
username: Annotated[
str, StringConstraints(pattern=r"^[a-zA-Z0-9_-]+$", min_length=3, max_length=36)
]
created_at: Optional[datetime] = Field(None, serialization_alias="createdAt")
status: Optional[UserStatus] = None
subscription_uuid: Optional[str] = Field(
None, serialization_alias="subscriptionUuid"
)
short_uuid: Optional[str] = Field(None, serialization_alias="shortUuid")
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[
Optional[str], StringConstraints(min_length=8, max_length=32)
str | None, StringConstraints(min_length=8, max_length=32)
] = Field(None, serialization_alias="trojanPassword")
vless_uuid: Optional[str] = Field(None, serialization_alias="vlessUuid")
vless_uuid: str | None = Field(None, serialization_alias="vlessUuid")
ss_password: Annotated[
Optional[str], StringConstraints(min_length=8, max_length=32)
str | None, StringConstraints(min_length=8, max_length=32)
] = Field(None, serialization_alias="ssPassword")
traffic_limit_bytes: Optional[int] = Field(
traffic_limit_bytes: int | None = Field(
None, serialization_alias="trafficLimitBytes", strict=True, ge=0
)
traffic_limit_strategy: Optional[TrafficLimitStrategy] = Field(
traffic_limit_strategy: TrafficLimitStrategy | None = Field(
None, serialization_alias="trafficLimitStrategy"
)
last_traffic_reset_at: Optional[datetime] = Field(
last_traffic_reset_at: datetime | None = Field(
None, serialization_alias="lastTrafficResetAt"
)
description: Optional[str] = None
tag: Optional[str] = None
telegram_id: Optional[int] = Field(None, serialization_alias="telegramId")
email: Optional[str] = None
hwidDeviceLimit: Optional[int] = Field(
description: str | None = None
tag: str | None = None
telegram_id: int | None = Field(None, serialization_alias="telegramId")
email: str | None = None
hwidDeviceLimit: int | None = Field(
None, serialization_alias="hwidDeviceLimit", strict=True, ge=0
)
active_internal_squads: Optional[List[str]] = Field(
active_internal_squads: list[str] | None = Field(
None, serialization_alias="activeInternalSquads"
)
class UpdateUserRequestDto(BaseModel):
uuid: UUID
active_internal_squads: Optional[List[str]] = Field(
active_internal_squads: list[str] | None = Field(
None, serialization_alias="activeInternalSquads"
)
description: Optional[str] = None
email: Optional[str] = None
expire_at: Optional[datetime] = Field(None, serialization_alias="expireAt")
hwidDeviceLimit: Optional[int] = Field(
description: str | None = None
email: str | None = None
expire_at: datetime | None = Field(None, serialization_alias="expireAt")
hwidDeviceLimit: int | None = Field(
None, serialization_alias="hwidDeviceLimit", strict=True, ge=0
)
status: Optional[UserStatus] = None
tag: Optional[str] = None
telegram_id: Optional[int] = Field(None, serialization_alias="telegramId")
traffic_limit_bytes: Optional[int] = 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: Optional[TrafficLimitStrategy] = Field(
traffic_limit_strategy: TrafficLimitStrategy | None = Field(
None, serialization_alias="trafficLimitStrategy"
)
class UserResponseDto(BaseModel):
uuid: UUID
subscription_uuid: Optional[UUID] = Field(None, alias="subscriptionUuid")
short_uuid: str = Field(alias="shortUuid")
username: str
status: Optional[UserStatus] = None
status: UserStatus | None = None
used_traffic_bytes: float = Field(alias="usedTrafficBytes")
lifetime_used_traffic_bytes: float = Field(alias="lifetimeUsedTrafficBytes")
traffic_limit_bytes: Optional[int] = Field(None, alias="trafficLimitBytes")
traffic_limit_strategy: Optional[str] = Field(None, alias="trafficLimitStrategy")
sub_last_user_agent: Optional[str] = Field(None, alias="subLastUserAgent")
sub_last_opened_at: Optional[datetime] = Field(None, alias="subLastOpenedAt")
expire_at: Optional[datetime] = Field(None, alias="expireAt")
online_at: Optional[datetime] = Field(None, alias="onlineAt")
sub_revoked_at: Optional[datetime] = Field(None, alias="subRevokedAt")
last_traffic_reset_at: Optional[datetime] = Field(None, alias="lastTrafficResetAt")
traffic_limit_bytes: int | None = Field(None, alias="trafficLimitBytes")
traffic_limit_strategy: str | None = Field(None, alias="trafficLimitStrategy")
sub_last_user_agent: str | None = Field(None, alias="subLastUserAgent")
sub_last_opened_at: datetime | None = Field(None, alias="subLastOpenedAt")
expire_at: datetime | None = Field(None, alias="expireAt")
online_at: datetime | None = Field(None, alias="onlineAt")
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")
description: Optional[str] = None
telegram_id: Optional[int] = Field(None, alias="telegramId")
email: Optional[str] = None
hwidDeviceLimit: Optional[int] = Field(
description: str | None = None
telegram_id: int | None = Field(None, alias="telegramId")
email: str | None = None
hwidDeviceLimit: int | None = Field(
None, serialization_alias="hwidDeviceLimit", strict=True, ge=0
)
active_internal_squads: Optional[List[ActiveInternalSquadDto]] = Field(
active_internal_squads: list[ActiveInternalSquadDto] | None = Field(
None, alias="activeInternalSquads"
)
subscription_url: str = Field(alias="subscriptionUrl")
first_connected: Optional[datetime] = Field(None, alias="firstConnectedAt")
last_trigger_threshold: Optional[int] = Field(None, alias="lastTriggeredThreshold")
last_connected_node: Optional[UserLastConnectedNodeDto] = Field(
subscription_url: str | None = Field(None, alias="subscriptionUrl")
first_connected: datetime | None = Field(None, alias="firstConnectedAt")
last_trigger_threshold: int | None = Field(None, alias="lastTriggeredThreshold")
last_connected_node: UserLastConnectedNodeDto | None = Field(
None, alias="lastConnectedNode"
)
happ: Optional[HappCrypto] = Field(None, alias="happ")
tag: Optional[str] = Field(None, alias="tag")
happ: HappCrypto | None = Field(None, alias="happ")
tag: str | None = Field(None, alias="tag")
created_at: datetime = Field(alias="createdAt")
updated_at: datetime = Field(alias="updatedAt")
class EmailUserResponseDto(RootModel[List[UserResponseDto]]):
class EmailUserResponseDto(RootModel[list[UserResponseDto]]):
def __iter__(self):
return iter(self.root)
@ -143,7 +139,7 @@ class EmailUserResponseDto(RootModel[List[UserResponseDto]]):
return self.root[item]
class TagUserResponseDto(RootModel[List[UserResponseDto]]):
class TagUserResponseDto(RootModel[list[UserResponseDto]]):
def __iter__(self):
return iter(self.root)
@ -151,7 +147,7 @@ class TagUserResponseDto(RootModel[List[UserResponseDto]]):
return self.root[item]
class TelegramUserResponseDto(RootModel[List[UserResponseDto]]):
class TelegramUserResponseDto(RootModel[list[UserResponseDto]]):
def __iter__(self):
return iter(self.root)
@ -160,7 +156,7 @@ class TelegramUserResponseDto(RootModel[List[UserResponseDto]]):
class UsersResponseDto(BaseModel):
users: List[UserResponseDto]
users: list[UserResponseDto]
total: float
@ -169,7 +165,7 @@ class DeleteUserResponseDto(BaseModel):
class TagsResponseDto(BaseModel):
tags: List[str]
tags: list[str]
class CreateUserResponseDto(UserResponseDto):
@ -213,7 +209,7 @@ class GetUserByUsernameResponseDto(UserResponseDto):
class RevokeUserRequestDto(BaseModel):
short_uuid: Optional[str] = Field(
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.",

53
tests/test_hwid.py
View file

@ -6,9 +6,11 @@ import pytest
from remnawave.models import (
CreateUserHwidDeviceRequestDto,
DeleteUserHwidDeviceRequestDto,
DeleteUserAllHwidDeviceRequestDto,
CreateUserHwidDeviceResponseDto,
DeleteUserHwidDeviceResponseDto,
GetUserHwidDevicesResponseDto,
GetHwidStatisticsResponseDto,
)
from tests.conftest import REMNAWAVE_USER_UUID
@ -21,6 +23,26 @@ async def test_get_hwid_user(remnawave):
assert hwid.devices is not None
@pytest.mark.asyncio
async def test_get_hwid_users(remnawave):
response = await remnawave.hwid.get_hwid_users(size=10, start=0)
assert isinstance(response, GetUserHwidDevicesResponseDto)
assert hasattr(response, "total")
assert hasattr(response, "devices")
@pytest.mark.asyncio
async def test_get_hwid_stats(remnawave):
response = await remnawave.hwid.get_hwid_stats()
assert isinstance(response, GetHwidStatisticsResponseDto)
assert hasattr(response, "by_platform")
assert hasattr(response, "by_app")
assert hasattr(response, "stats")
assert hasattr(response.stats, "total_unique_devices")
assert hasattr(response.stats, "total_hwid_devices")
assert hasattr(response.stats, "average_hwid_devices_per_user")
@pytest.mark.asyncio
async def test_add_hwid_to_user(remnawave):
create_request = CreateUserHwidDeviceRequestDto(
@ -28,8 +50,8 @@ async def test_add_hwid_to_user(remnawave):
user_uuid=REMNAWAVE_USER_UUID,
platform="Windows",
os_version="10.0.19042",
deviceModel="Surface Pro",
userAgent="Mozilla/5.0"
device_model="Surface Pro",
user_agent="Mozilla/5.0"
)
response = await remnawave.hwid.add_hwid_to_users(body=create_request)
assert isinstance(response, CreateUserHwidDeviceResponseDto)
@ -45,3 +67,30 @@ async def test_delete_hwid_user(remnawave):
response = await remnawave.hwid.delete_hwid_to_user(body=delete_request)
assert isinstance(response, DeleteUserHwidDeviceResponseDto)
assert not any(item.hwid == new_hwid for item in response.devices)
@pytest.mark.asyncio
async def test_delete_all_hwid_user(remnawave):
# Сначала добавим новый HWID
create_request = CreateUserHwidDeviceRequestDto(
hwid=str(uuid.uuid4()),
user_uuid=REMNAWAVE_USER_UUID,
platform="iOS",
os_version="15.0",
device_model="iPhone 13",
user_agent="Safari/605.1.15"
)
await remnawave.hwid.add_hwid_to_users(body=create_request)
# Теперь удалим все HWID устройства пользователя
delete_all_request = DeleteUserAllHwidDeviceRequestDto(
user_uuid=REMNAWAVE_USER_UUID
)
response = await remnawave.hwid.delete_all_hwid_user(body=delete_all_request)
assert isinstance(response, DeleteUserHwidDeviceResponseDto)
assert len(response.devices) == 0
# Проверим, что устройства действительно удалены
hwid_check = await remnawave.hwid.get_hwid_user(uuid=REMNAWAVE_USER_UUID)
assert len(hwid_check.devices) == 0

14
tests/test_users.py
View file

@ -62,13 +62,13 @@ async def test_users(remnawave) -> None:
assert user_short_uuid.uuid == create_user.uuid
# Only test get_user_by_subscription_uuid if subscription_uuid is not None
if create_user.subscription_uuid is not None:
string_subscription_uuid = str(create_user.subscription_uuid)
user_subscription_uuid = await remnawave.users.get_user_by_subscription_uuid(
subscription_uuid=string_subscription_uuid
)
assert isinstance(user_subscription_uuid, UserResponseDto)
assert user_subscription_uuid.uuid == create_user.uuid
# if create_user.subscription_uuid is not None:
# string_subscription_uuid = str(create_user.subscription_uuid)
# user_subscription_uuid = await remnawave.users.get_user_by_subscription_uuid(
# subscription_uuid=string_subscription_uuid
# )
# assert isinstance(user_subscription_uuid, UserResponseDto)
# assert user_subscription_uuid.uuid == create_user.uuid
user_username = await remnawave.users.get_user_by_username(
username=user_uuid.username