You've already forked wc-licensed-product
234 lines
7.0 KiB
Python
234 lines
7.0 KiB
Python
|
|
"""
|
||
|
|
WC Licensed Product API Client for Python
|
||
|
|
|
||
|
|
A simple Python client for interacting with the WC Licensed Product REST API.
|
||
|
|
|
||
|
|
Requirements:
|
||
|
|
- Python 3.7+
|
||
|
|
- requests library (pip install requests)
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
client = WcLicensedProductClient('https://your-site.com')
|
||
|
|
result = client.validate('XXXX-XXXX-XXXX-XXXX', 'example.com')
|
||
|
|
"""
|
||
|
|
|
||
|
|
import requests
|
||
|
|
from typing import Dict, Any, Optional
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class LicenseStatus:
|
||
|
|
"""License status response data."""
|
||
|
|
valid: bool
|
||
|
|
status: str
|
||
|
|
domain: Optional[str]
|
||
|
|
expires_at: Optional[str]
|
||
|
|
activations_count: int
|
||
|
|
max_activations: int
|
||
|
|
|
||
|
|
|
||
|
|
class WcLicensedProductError(Exception):
|
||
|
|
"""Base exception for API errors."""
|
||
|
|
def __init__(self, message: str, error_code: str = None, http_code: int = None):
|
||
|
|
super().__init__(message)
|
||
|
|
self.error_code = error_code
|
||
|
|
self.http_code = http_code
|
||
|
|
|
||
|
|
|
||
|
|
class RateLimitError(WcLicensedProductError):
|
||
|
|
"""Raised when rate limit is exceeded."""
|
||
|
|
def __init__(self, retry_after: int):
|
||
|
|
super().__init__(f"Rate limit exceeded. Retry after {retry_after} seconds.")
|
||
|
|
self.retry_after = retry_after
|
||
|
|
|
||
|
|
|
||
|
|
class WcLicensedProductClient:
|
||
|
|
"""Client for the WC Licensed Product REST API."""
|
||
|
|
|
||
|
|
def __init__(self, site_url: str, timeout: int = 30):
|
||
|
|
"""
|
||
|
|
Initialize the client.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
site_url: Your WordPress site URL (e.g., 'https://example.com')
|
||
|
|
timeout: Request timeout in seconds
|
||
|
|
"""
|
||
|
|
self.base_url = site_url.rstrip('/') + '/wp-json/wc-licensed-product/v1'
|
||
|
|
self.timeout = timeout
|
||
|
|
self.session = requests.Session()
|
||
|
|
self.session.headers.update({
|
||
|
|
'Content-Type': 'application/json',
|
||
|
|
'Accept': 'application/json',
|
||
|
|
})
|
||
|
|
|
||
|
|
def validate(self, license_key: str, domain: str) -> Dict[str, Any]:
|
||
|
|
"""
|
||
|
|
Validate a license key for a specific domain.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
license_key: The license key to validate
|
||
|
|
domain: The domain to validate against
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Dict with 'valid' boolean and additional license info
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
WcLicensedProductError: On API error
|
||
|
|
RateLimitError: When rate limited
|
||
|
|
"""
|
||
|
|
return self._request('/validate', {
|
||
|
|
'license_key': license_key,
|
||
|
|
'domain': domain,
|
||
|
|
})
|
||
|
|
|
||
|
|
def status(self, license_key: str) -> LicenseStatus:
|
||
|
|
"""
|
||
|
|
Get the status of a license.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
license_key: The license key to check
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
LicenseStatus object with license details
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
WcLicensedProductError: On API error
|
||
|
|
RateLimitError: When rate limited
|
||
|
|
"""
|
||
|
|
data = self._request('/status', {
|
||
|
|
'license_key': license_key,
|
||
|
|
})
|
||
|
|
|
||
|
|
return LicenseStatus(
|
||
|
|
valid=data.get('valid', False),
|
||
|
|
status=data.get('status', 'unknown'),
|
||
|
|
domain=data.get('domain'),
|
||
|
|
expires_at=data.get('expires_at'),
|
||
|
|
activations_count=data.get('activations_count', 0),
|
||
|
|
max_activations=data.get('max_activations', 1),
|
||
|
|
)
|
||
|
|
|
||
|
|
def activate(self, license_key: str, domain: str) -> Dict[str, Any]:
|
||
|
|
"""
|
||
|
|
Activate a license on a domain.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
license_key: The license key to activate
|
||
|
|
domain: The domain to activate on
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Dict with 'success' boolean and message
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
WcLicensedProductError: On API error
|
||
|
|
RateLimitError: When rate limited
|
||
|
|
"""
|
||
|
|
return self._request('/activate', {
|
||
|
|
'license_key': license_key,
|
||
|
|
'domain': domain,
|
||
|
|
})
|
||
|
|
|
||
|
|
def _request(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
|
|
"""
|
||
|
|
Make an API request.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
endpoint: API endpoint path
|
||
|
|
data: Request data
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Decoded response data
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
WcLicensedProductError: On API error
|
||
|
|
RateLimitError: When rate limited
|
||
|
|
"""
|
||
|
|
url = self.base_url + endpoint
|
||
|
|
|
||
|
|
try:
|
||
|
|
response = self.session.post(url, json=data, timeout=self.timeout)
|
||
|
|
except requests.RequestException as e:
|
||
|
|
raise WcLicensedProductError(f"Request failed: {e}")
|
||
|
|
|
||
|
|
# Handle rate limiting
|
||
|
|
if response.status_code == 429:
|
||
|
|
retry_after = int(response.headers.get('Retry-After', 60))
|
||
|
|
raise RateLimitError(retry_after)
|
||
|
|
|
||
|
|
try:
|
||
|
|
result = response.json()
|
||
|
|
except ValueError:
|
||
|
|
raise WcLicensedProductError("Invalid JSON response")
|
||
|
|
|
||
|
|
# Check for API errors
|
||
|
|
if response.status_code >= 400 and 'error' in result:
|
||
|
|
raise WcLicensedProductError(
|
||
|
|
result.get('message', 'Unknown error'),
|
||
|
|
error_code=result.get('error'),
|
||
|
|
http_code=response.status_code,
|
||
|
|
)
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
# =============================================================================
|
||
|
|
# Usage Examples
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
# Initialize the client
|
||
|
|
client = WcLicensedProductClient('https://your-wordpress-site.com')
|
||
|
|
|
||
|
|
# Example 1: Validate a license
|
||
|
|
print("=== Validate License ===")
|
||
|
|
try:
|
||
|
|
result = client.validate('XXXX-XXXX-XXXX-XXXX', 'myapp.example.com')
|
||
|
|
|
||
|
|
if result.get('valid'):
|
||
|
|
print("License is valid!")
|
||
|
|
print(f"Status: {result.get('license', {}).get('status', 'active')}")
|
||
|
|
print(f"Expires: {result.get('license', {}).get('expires_at', 'Never')}")
|
||
|
|
else:
|
||
|
|
print(f"License is not valid: {result.get('message', 'Unknown error')}")
|
||
|
|
|
||
|
|
except RateLimitError as e:
|
||
|
|
print(f"Rate limited. Retry after {e.retry_after} seconds.")
|
||
|
|
except WcLicensedProductError as e:
|
||
|
|
print(f"Error: {e}")
|
||
|
|
|
||
|
|
print()
|
||
|
|
|
||
|
|
# Example 2: Check license status
|
||
|
|
print("=== Check License Status ===")
|
||
|
|
try:
|
||
|
|
status = client.status('XXXX-XXXX-XXXX-XXXX')
|
||
|
|
|
||
|
|
print(f"Valid: {'Yes' if status.valid else 'No'}")
|
||
|
|
print(f"Status: {status.status}")
|
||
|
|
print(f"Domain: {status.domain or 'None'}")
|
||
|
|
print(f"Expires: {status.expires_at or 'Never'}")
|
||
|
|
print(f"Activations: {status.activations_count}/{status.max_activations}")
|
||
|
|
|
||
|
|
except RateLimitError as e:
|
||
|
|
print(f"Rate limited. Retry after {e.retry_after} seconds.")
|
||
|
|
except WcLicensedProductError as e:
|
||
|
|
print(f"Error: {e}")
|
||
|
|
|
||
|
|
print()
|
||
|
|
|
||
|
|
# Example 3: Activate license on a domain
|
||
|
|
print("=== Activate License ===")
|
||
|
|
try:
|
||
|
|
result = client.activate('XXXX-XXXX-XXXX-XXXX', 'newsite.example.com')
|
||
|
|
|
||
|
|
if result.get('success'):
|
||
|
|
print("License activated successfully!")
|
||
|
|
else:
|
||
|
|
print(f"Activation failed: {result.get('message', 'Unknown error')}")
|
||
|
|
|
||
|
|
except RateLimitError as e:
|
||
|
|
print(f"Rate limited. Retry after {e.retry_after} seconds.")
|
||
|
|
except WcLicensedProductError as e:
|
||
|
|
print(f"Error ({e.error_code}): {e}")
|