Skip to content

Commit

Permalink
refactor: update all methods and tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@orenlab
orenlab committed Sep 1, 2024
1 parent e9786f3 commit 4716799
Show file tree
Hide file tree
Showing 4 changed files with 197 additions and 324 deletions.
250 changes: 92 additions & 158 deletions pyoutlineapi/client.py
View file
Original file line number Diff line number Diff line change
@@ -1,51 +1,39 @@
from typing import Optional, Type, Any
from typing import Union, Optional, Type

import requests
from pydantic import BaseModel, ValidationError as PydanticValidationError
from pydantic import BaseModel, ValidationError
from requests_toolbelt.adapters.fingerprint import FingerprintAdapter

from pyoutlineapi.exceptions import APIError, ValidationError, HTTPError
from pyoutlineapi.logger import setup_logger
from pyoutlineapi.models import (
AccessKeyCreateRequest,
Server,
AccessKey,
AccessKeyList,
ServerPort,
DataLimit,
Metrics
)

# Set up logging
logger = setup_logger(__name__)

__all__ = [
'PyOutlineWrapper'
]
from pyoutlineapi.exceptions import APIError
from pyoutlineapi.models import DataLimit, ServerPort, Metrics, AccessKeyList, AccessKey, AccessKeyCreateRequest, Server


class PyOutlineWrapper:
"""
Class for interacting with the Outline VPN Server.
Class for interacting with the Outline VPN Server API.
This class provides methods to interact with the Outline VPN Server, including:
This class provides methods for managing access keys, retrieving server
information, updating server settings, and monitoring data usage.
- Retrieving server information
- Creating, listing, and deleting access keys
- Updating server ports
- Setting and removing data limits for access keys
- Retrieving metrics
Attributes:
_api_url (str): The base URL of the API.
_cert_sha256 (str): SHA-256 fingerprint of the certificate for authenticity verification.
_verify_tls (bool): Whether to verify the TLS certificate.
The class uses the `requests` library for making HTTP requests and `pydantic` for data validation.
Responses can be returned either as Pydantic models or in JSON format, depending on the `json_format` parameter.
"""

def __init__(self, api_url: str, cert_sha256: str, verify_tls: bool = True, json_format: Optional[bool] = False):
def __init__(self, api_url: str, cert_sha256: str, verify_tls: bool = True, json_format: bool = True):
"""
Initialize PyOutlineAPI.
Initializes the PyOutlineWrapper with the given API URL, certificate fingerprint, and options for TLS verification
and response format.
Args:
api_url (str): The base URL of the API.
cert_sha256 (str): SHA-256 fingerprint of the certificate.
verify_tls (bool, optional): Whether to verify the TLS certificate. Defaults to True.
json_format (bool, optional): Whether to return responses in JSON format. Defaults to False.
api_url (str): The base URL of the Outline VPN Server API.
cert_sha256 (str): The SHA-256 fingerprint of the server's certificate.
verify_tls (bool, optional): Whether to verify the server's TLS certificate. Defaults to True.
json_format (bool, optional): Whether to return responses in JSON format. Defaults to True.
"""
self._api_url = api_url
self._cert_sha256 = cert_sha256
Expand All @@ -56,18 +44,18 @@ def __init__(self, api_url: str, cert_sha256: str, verify_tls: bool = True, json

def _request(self, method: str, endpoint: str, json_data=None) -> requests.Response:
"""
Perform an HTTP request to the API.
Makes an HTTP request to the API.
Args:
method (str): HTTP method (GET, POST, PUT, DELETE).
endpoint (str): API endpoint.
json_data (Dict[str, Any], optional): Data to send in the request body.
method (str): The HTTP method to use (e.g., 'GET', 'POST', 'PUT', 'DELETE').
endpoint (str): The API endpoint to call.
json_data (optional): The JSON data to send with the request.
Returns:
requests.Response: The HTTP response object.
Raises:
APIError: If the request fails.
APIError: If the request fails or the response status is not successful.
"""
url = f"{self._api_url}/{endpoint}"
try:
Expand All @@ -80,69 +68,52 @@ def _request(self, method: str, endpoint: str, json_data=None) -> requests.Respo
)
response.raise_for_status()
return response
except (requests.RequestException, HTTPError) as exception:
except requests.RequestException as exception:
raise APIError(f"Request to {url} failed: {exception}")

def _parse_response(self, response: requests.Response, model: Type[BaseModel]) -> Any:
def _parse_response(self, response: requests.Response, model: Type[BaseModel]) -> Union[BaseModel, str]:
"""
Validate response data and optionally convert to JSON format.
Parses the response from the API.
Args:
response (requests.Response): The HTTP response object.
model (Type[BaseModel]): The Pydantic model class for validation.
model (Type[BaseModel]): The Pydantic model to validate the response data.
Returns:
Any: The validated and optionally JSON-formatted response data.
Union[BaseModel, str]: The validated data as a Pydantic model or a JSON string, depending on the json_format parameter.
Raises:
ValidationError: If the response data is invalid.
ValidationError: If the response data does not match the Pydantic model.
"""
try:
data = model(**response.json())
if self._json_format:
return data.model_dump_json()
return data
except PydanticValidationError as e:
json_data = response.json()
data = model.model_validate(json_data)
return data.model_dump_json() if self._json_format else data
except ValidationError as e:
raise ValidationError(f"Validation error: {e}")

def get_server_info(self) -> Any:
def get_server_info(self) -> Union[Server, str]:
"""
Get information about the server.
Retrieves information about the Outline VPN server.
Returns:
Any: An object containing server information, or JSON-formatted data if json_format is True.
Raises:
APIError: If the request fails.
ValidationError: If the response data is invalid.
Example:
server_info = api.get_server_info()
Union[Server, str]: The server information as a Pydantic model or a JSON string.
"""
try:
response = self._request("GET", "server")
return self._parse_response(response, Server)
except PydanticValidationError as e:
raise ValidationError(f"Failed to get server info: {e}")
response = self._request("GET", "server")
return self._parse_response(response, Server)

def create_access_key(self, name: Optional[str] = None, password: Optional[str] = None,
port: Optional[int] = None) -> Any:
port: Optional[int] = None) -> Union[AccessKey, str]:
"""
Create a new access key.
Creates a new access key.
Args:
name (Optional[str]): Name of the access key. Defaults to None.
password (Optional[str]): Password for the access key. Defaults to None.
port (Optional[int]): Port for the access key. Defaults to None.
name (Optional[str]): The name of the access key.
password (Optional[str]): The password for the access key.
port (Optional[int]): The port for the access key.
Returns:
Any: An object containing information about the new access key, or JSON-formatted data if json_format is True.
Raises:
ValidationError: If the server response is not 201 or if there's an issue with the request.
Example:
access_key = api.create_access_key(name="User1", password="securepassword")
Union[AccessKey, str]: The created access key as a Pydantic model or a JSON string.
"""
request_data = {
"name": name,
Expand All @@ -151,127 +122,90 @@ def create_access_key(self, name: Optional[str] = None, password: Optional[str]
}
request_data = {key: value for key, value in request_data.items() if value is not None}

try:
if request_data:
request_data = AccessKeyCreateRequest(**request_data).model_dump(mode="json")
else:
request_data = None

response = self._request("POST", "access-keys", json_data=request_data)
if request_data:
request_data = AccessKeyCreateRequest(**request_data).model_dump()

if response.status_code == 201:
return self._parse_response(response, AccessKey)
else:
raise ValidationError(f"Failed to create access key: {response.status_code} - {response.text}")
response = self._request("POST", "access-keys", json_data=request_data)
return self._parse_response(response, AccessKey)

except (PydanticValidationError, KeyError, requests.RequestException) as e:
raise ValidationError(f"Failed to create access key: {e}")

def get_access_keys(self) -> Any:
def get_access_keys(self) -> Union[AccessKeyList, str]:
"""
Get a list of all access keys.
Retrieves a list of all access keys.
Returns:
Any: An object containing a list of access keys, or JSON-formatted data if json_format is True.
Raises:
APIError: If the request fails.
ValidationError: If the response data is invalid.
Example:
access_keys = api.get_access_keys()
Union[AccessKeyList, str]: The list of access keys as a Pydantic model or a JSON string.
"""
try:
response = self._request("GET", "access-keys")
return self._parse_response(response, AccessKeyList)
except PydanticValidationError as e:
raise ValidationError(f"Failed to get access keys: {e}")
response = self._request("GET", "access-keys")
return self._parse_response(response, AccessKeyList)

def delete_access_key(self, key_id: str) -> bool:
"""
Delete an access key by its ID.
Deletes an access key by its ID.
Args:
key_id (str): The ID of the access key.
key_id (str): The ID of the access key to delete.
Returns:
bool: True if the access key was successfully deleted, False otherwise.
Raises:
APIError: If the request fails.
Example:
success = api.delete_access_key(key_id="some_key_id")
"""
try:
response = self._request("DELETE", f"access-keys/{key_id}")
return response.status_code == 204
except HTTPError as e:
raise APIError(f"Failed to delete access key with ID {key_id}: {e}")
response = self._request("DELETE", f"access-keys/{key_id}")
return response.status_code == 204

def update_server_port(self, port: ServerPort | int) -> bool:
def update_server_port(self, port: int) -> bool:
"""
Update the port for new access keys.
Updates the port for new access keys on the server.
Args:
port (ServerPort): The new port.
port (int): The new port number.
Returns:
bool: True if the server port was successfully updated, False otherwise.
bool: True if the port was successfully updated, False otherwise.
Raises:
APIError: If the request fails.
Example:
success = api.update_server_port(port=12345)
APIError: If the port is already in use.
"""
try:
response = self._request("PUT", "server/port-for-new-access-keys", {"port": port})
if response.status_code == 409:
raise APIError(f"Port {port} is already in use")
return response.status_code == 204
except (PydanticValidationError, APIError) as e:
raise APIError(f"Failed to update server port: {e}")
verified_port = ServerPort(port=port)
response = self._request("PUT", "server/port-for-new-access-keys", {"port": verified_port.port})
if response.status_code == 409:
raise APIError(f"Port {verified_port.port} is already in use")
return response.status_code == 204

def set_access_key_data_limit(self, key_id: str, limit: DataLimit) -> bool:
"""
Set the data limit for an access key.
Sets a data limit for an access key.
Args:
key_id (str): The ID of the access key.
limit (DataLimit): The data limit in bytes.
limit (DataLimit): The data limit to set.
Returns:
bool: True if the data limit was successfully set, False otherwise.
"""
response = self._request("PUT", f"access-keys/{key_id}/data-limit", {"bytes": limit.bytes})
return response.status_code == 204

Raises:
APIError: If the request fails.
def get_metrics(self) -> Union[Metrics, str]:
"""
Retrieves transfer metrics from the server.
Example:
success = api.set_access_key_data_limit(key_id="some_key_id", limit=DataLimit(bytes=1048576))
Returns:
Union[Metrics, str]: The metrics data as a Pydantic model or a JSON string.
"""
try:
response = self._request("PUT", f"access-keys/{key_id}/data-limit", {"bytes": limit})
return response.status_code == 204
except (PydanticValidationError, APIError) as e:
raise APIError(f"Failed to set data limit for access key with ID {key_id}: {e}")
response = self._request("GET", "metrics/transfer")
return self._parse_response(response, Metrics)

def get_metrics(self) -> Metrics:
def remove_access_key_data_limit(self, key_id: str) -> bool:
"""
Get metrics about data transfer.
Removes the data limit for an access key.
Args:
key_id (str): The ID of the access key.
Returns:
Metrics: An object containing data transfer metrics.
bool: True if the data limit was successfully removed, False otherwise.
"""
response = self._request("DELETE", f"access-keys/{key_id}/data-limit")
return response.status_code == 204

Raises:
APIError: If the request fails.
ValidationError: If the response data is invalid.

Example:
metrics = api.get_metrics()
"""
try:
response = self._request("GET", "metrics/transfer")
return self._parse_response(response, Metrics)
except PydanticValidationError as e:
raise ValidationError(f"Failed to get metrics: {e}")
__all__ = ["PyOutlineWrapper"]
10 changes: 5 additions & 5 deletions pyoutlineapi/models.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

from typing import Optional, List, Dict

from pydantic import BaseModel, Field, SecretStr, constr, field_validator
from pydantic import BaseModel, Field, constr, field_validator


class Server(BaseModel):
Expand Down Expand Up @@ -48,17 +48,17 @@ class AccessKey(BaseModel):
Attributes:
id (str): The unique identifier for the access key.
name (str): The name of the access key.
password (SecretStr): The password for the access key, must not be empty.
password (str): The password for the access key, must not be empty.
port (int): The port used by the access key, must be between 1 and 65535.
method (str): The encryption method used by the access key.
accessUrl (SecretStr): The URL used to access the server, must not be empty.
accessUrl (str): The URL used to access the server, must not be empty.
"""
id: str
name: str
password: SecretStr = Field(..., min_length=1, description="Password must not be empty")
password: str = Field(..., min_length=1, description="Password must not be empty")
port: int = Field(ge=1, le=65535, description="Port must be between 1 and 65535")
method: str
accessUrl: SecretStr = Field(..., min_length=1, description="Access URL must not be empty")
accessUrl: str = Field(..., min_length=1, description="Access URL must not be empty")


class ServerPort(BaseModel):
Expand Down
Loading

0 comments on commit 4716799

Please sign in to comment.