From 21cd3f004a6ef555b9d3fbed46f87889251bcf8d Mon Sep 17 00:00:00 2001 From: David Friedel Date: Thu, 25 Dec 2025 11:17:34 +0000 Subject: [PATCH] Implement IronLicensing Python SDK - LicenseClient for validation and activation - Feature checking with has_feature/require_feature - Trial management - Purchase/checkout flow - Offline caching with grace period - License change callbacks - Full async support with httpx - Comprehensive README --- .gitignore | 36 +++ README.md | 279 ++++++++++++++++++++- pyproject.toml | 38 +++ src/ironlicensing/__init__.py | 202 +++++++++++++++ src/ironlicensing/cache.py | 160 ++++++++++++ src/ironlicensing/client.py | 288 +++++++++++++++++++++ src/ironlicensing/transport.py | 440 +++++++++++++++++++++++++++++++++ src/ironlicensing/types.py | 113 +++++++++ 8 files changed, 1554 insertions(+), 2 deletions(-) create mode 100644 .gitignore create mode 100644 pyproject.toml create mode 100644 src/ironlicensing/__init__.py create mode 100644 src/ironlicensing/cache.py create mode 100644 src/ironlicensing/client.py create mode 100644 src/ironlicensing/transport.py create mode 100644 src/ironlicensing/types.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..359a6e2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,36 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +dist/ +build/ +*.egg-info/ +*.egg + +# Virtual environments +.venv/ +venv/ +ENV/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# OS +.DS_Store +Thumbs.db + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# Type checking +.mypy_cache/ + +# Local storage +.ironlicensing/ diff --git a/README.md b/README.md index 95a32bf..a3408b6 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,277 @@ -# ironlicensing-python -IronLicensing SDK for Python - Software licensing and activation +# IronLicensing SDK for Python + +Software licensing and activation SDK for Python applications. Validate licenses, manage activations, check features, and handle trials. + +[![PyPI](https://img.shields.io/pypi/v/ironlicensing.svg)](https://pypi.org/project/ironlicensing/) +[![Python](https://img.shields.io/pypi/pyversions/ironlicensing.svg)](https://pypi.org/project/ironlicensing/) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) + +## Installation + +```bash +pip install ironlicensing +``` + +## Quick Start + +### Validate a License + +```python +import ironlicensing + +# Initialize +ironlicensing.init("pk_live_xxxxx", "my-app") + +# Validate a license key +result = ironlicensing.validate("IRON-XXXX-XXXX-XXXX-XXXX") + +if result.valid: + print("License is valid!") + print(f"Expires: {result.license.expires_at}") +``` + +### Activate a License + +```python +import ironlicensing + +ironlicensing.init("pk_live_xxxxx", "my-app") + +# Activate on this machine +result = ironlicensing.activate("IRON-XXXX-XXXX-XXXX-XXXX") + +if result.valid: + print("License activated!") + print(f"Activations: {len(result.activations)}") +``` + +### Check Features + +```python +import ironlicensing +from ironlicensing import LicenseRequiredError + +ironlicensing.init("pk_live_xxxxx", "my-app") +ironlicensing.validate("IRON-XXXX-XXXX-XXXX-XXXX") + +# Check if feature is enabled +if ironlicensing.has_feature("premium"): + # Enable premium features + pass + +# Or require a feature (raises if not available) +try: + ironlicensing.require_feature("enterprise") + # Enterprise code here +except LicenseRequiredError: + print("Enterprise license required") +``` + +### Async Support + +```python +import asyncio +import ironlicensing + +async def main(): + ironlicensing.init("pk_live_xxxxx", "my-app") + + # Async validation + result = await ironlicensing.validate_async("IRON-XXXX-XXXX-XXXX-XXXX") + + # Async activation + result = await ironlicensing.activate_async("IRON-XXXX-XXXX-XXXX-XXXX") + + # Async trial + result = await ironlicensing.start_trial_async("user@example.com") + +asyncio.run(main()) +``` + +### Using the Client Directly + +```python +from ironlicensing import LicenseClient, LicenseOptions + +client = LicenseClient(LicenseOptions( + public_key="pk_live_xxxxx", + product_slug="my-app", + debug=True, +)) + +# Validate +result = client.validate("IRON-XXXX-XXXX-XXXX-XXXX") + +# Check features +if client.has_feature("premium"): + pass + +# Clean up +client.close() +``` + +## Configuration + +```python +from ironlicensing import LicenseClient, LicenseOptions + +client = LicenseClient(LicenseOptions( + public_key="pk_live_xxxxx", # Required + product_slug="my-app", # Required + api_base_url="https://api.ironlicensing.com", + debug=False, + enable_offline_cache=True, + cache_validation_minutes=60, + offline_grace_days=7, + storage_key_prefix="ironlicensing", +)) +``` + +### Configuration Options + +| Option | Type | Default | Description | +|--------|------|---------|-------------| +| `public_key` | str | required | Your public key (pk_live_xxx or pk_test_xxx) | +| `product_slug` | str | required | Your product slug | +| `api_base_url` | str | https://api.ironlicensing.com | API base URL | +| `debug` | bool | False | Enable debug logging | +| `enable_offline_cache` | bool | True | Cache validation results | +| `cache_validation_minutes` | int | 60 | Cache validity in minutes | +| `offline_grace_days` | int | 7 | Offline grace period in days | + +## License Status + +```python +LicenseStatus = Literal[ + "valid", # License is valid + "expired", # License has expired + "suspended", # License is suspended + "revoked", # License is revoked + "invalid", # License key is invalid + "trial", # Trial license + "trial_expired", # Trial has expired + "not_activated", # No license activated + "unknown", # Unknown status +] +``` + +## Features + +### Check Features + +```python +# Check if feature is enabled +if client.has_feature("premium"): + enable_premium_features() + +# Get feature details +feature = client.get_feature("max-users") +if feature and feature.enabled: + print(f"Max users: {feature.metadata.get('limit')}") + +# Get all features +for f in client.get_features(): + print(f"{f.name}: {'enabled' if f.enabled else 'disabled'}") +``` + +### Require Features + +```python +from ironlicensing import LicenseRequiredError + +try: + client.require_feature("enterprise") + # This code only runs if feature is available +except LicenseRequiredError: + show_upgrade_dialog() +``` + +## Trial Management + +```python +# Start a trial +result = ironlicensing.start_trial("user@example.com") + +if result.valid: + print("Trial started!") + print(f"Expires: {result.license.expires_at}") + +# Check if trial +if ironlicensing.is_trial(): + show_trial_banner(ironlicensing.get_license().expires_at) +``` + +## Purchase Flow + +```python +# Get available tiers +tiers = ironlicensing.get_tiers() + +for tier in tiers: + print(f"{tier.name}: {tier.price} {tier.currency}/{tier.billing_period}") + +# Start checkout +checkout = ironlicensing.start_purchase("tier-pro", "user@example.com") + +if checkout.success: + # Open checkout URL + print(f"Checkout URL: {checkout.checkout_url}") +``` + +## Deactivation + +```python +# Deactivate on this machine +if ironlicensing.deactivate(): + print("License deactivated") +``` + +## License Change Events + +```python +def on_license_change(license): + if license: + print(f"License updated: {license.status}") + else: + print("License removed") + +# Listen for license changes +unsubscribe = client.on_license_change(on_license_change) + +# Later: unsubscribe +unsubscribe() +``` + +## Offline Support + +The SDK automatically caches validation results and supports offline usage: + +```python +client = LicenseClient(LicenseOptions( + public_key="pk_live_xxxxx", + product_slug="my-app", + enable_offline_cache=True, + cache_validation_minutes=60, # Use cache for 60 minutes + offline_grace_days=7, # Allow 7 days offline +)) + +# This will use cache if available +result = client.validate() + +if result.cached: + print("Using cached validation") +``` + +## Requirements + +- Python 3.8+ +- httpx + +## Links + +- [Documentation](https://www.ironlicensing.com/docs) +- [Dashboard](https://www.ironlicensing.com) + +## License + +MIT License - see [LICENSE](LICENSE) for details. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..819fa73 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,38 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "ironlicensing" +version = "1.0.0" +description = "Software licensing and activation SDK for Python" +readme = "README.md" +license = "MIT" +authors = [ + { name = "IronServices", email = "support@ironservices.com" } +] +keywords = ["licensing", "activation", "license-key", "software-licensing", "feature-flags"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Typing :: Typed", +] +requires-python = ">=3.8" +dependencies = [ + "httpx>=0.24.0", +] + +[project.urls] +Homepage = "https://www.ironlicensing.com" +Documentation = "https://www.ironlicensing.com/docs" +Repository = "https://github.com/IronServices/ironlicensing-python" + +[tool.hatch.build.targets.wheel] +packages = ["src/ironlicensing"] diff --git a/src/ironlicensing/__init__.py b/src/ironlicensing/__init__.py new file mode 100644 index 0000000..80fef4f --- /dev/null +++ b/src/ironlicensing/__init__.py @@ -0,0 +1,202 @@ +"""IronLicensing SDK for Python. + +Software licensing and activation SDK. + +Example: + >>> import ironlicensing + >>> + >>> # Initialize + >>> ironlicensing.init("pk_live_xxxxx", "my-app") + >>> + >>> # Validate a license + >>> result = ironlicensing.validate("IRON-XXXX-XXXX-XXXX-XXXX") + >>> if result.valid: + ... print("License is valid!") + >>> + >>> # Check features + >>> if ironlicensing.has_feature("premium"): + ... print("Premium features enabled") +""" + +from .types import ( + LicenseStatus, + LicenseType, + LicenseOptions, + Feature, + License, + Activation, + LicenseResult, + CheckoutResult, + ProductTier, +) +from .client import LicenseClient, LicenseRequiredError + +__all__ = [ + # Types + "LicenseStatus", + "LicenseType", + "LicenseOptions", + "Feature", + "License", + "Activation", + "LicenseResult", + "CheckoutResult", + "ProductTier", + # Client + "LicenseClient", + "LicenseRequiredError", + # Global functions + "init", + "validate", + "validate_async", + "activate", + "activate_async", + "deactivate", + "deactivate_async", + "start_trial", + "start_trial_async", + "has_feature", + "require_feature", + "get_feature", + "get_features", + "get_license", + "get_status", + "is_licensed", + "is_trial", + "get_tiers", + "get_tiers_async", + "start_purchase", + "start_purchase_async", +] + +_global_client: LicenseClient | None = None + + +def init( + public_key: str, + product_slug: str, + api_base_url: str = "https://api.ironlicensing.com", + debug: bool = False, + enable_offline_cache: bool = True, + cache_validation_minutes: int = 60, + offline_grace_days: int = 7, +) -> LicenseClient: + """Initialize the global license client.""" + global _global_client + options = LicenseOptions( + public_key=public_key, + product_slug=product_slug, + api_base_url=api_base_url, + debug=debug, + enable_offline_cache=enable_offline_cache, + cache_validation_minutes=cache_validation_minutes, + offline_grace_days=offline_grace_days, + ) + _global_client = LicenseClient(options) + return _global_client + + +def _get_client() -> LicenseClient: + if _global_client is None: + raise RuntimeError("IronLicensing not initialized. Call init() first.") + return _global_client + + +def validate(license_key: str | None = None) -> LicenseResult: + """Validate a license key (sync).""" + return _get_client().validate(license_key) + + +async def validate_async(license_key: str | None = None) -> LicenseResult: + """Validate a license key (async).""" + return await _get_client().validate_async(license_key) + + +def activate(license_key: str, machine_name: str | None = None) -> LicenseResult: + """Activate a license (sync).""" + return _get_client().activate(license_key, machine_name) + + +async def activate_async(license_key: str, machine_name: str | None = None) -> LicenseResult: + """Activate a license (async).""" + return await _get_client().activate_async(license_key, machine_name) + + +def deactivate() -> bool: + """Deactivate the current license (sync).""" + return _get_client().deactivate() + + +async def deactivate_async() -> bool: + """Deactivate the current license (async).""" + return await _get_client().deactivate_async() + + +def start_trial(email: str) -> LicenseResult: + """Start a trial (sync).""" + return _get_client().start_trial(email) + + +async def start_trial_async(email: str) -> LicenseResult: + """Start a trial (async).""" + return await _get_client().start_trial_async(email) + + +def has_feature(feature_key: str) -> bool: + """Check if a feature is enabled.""" + return _get_client().has_feature(feature_key) + + +def require_feature(feature_key: str) -> None: + """Require a feature, raises LicenseRequiredError if not available.""" + _get_client().require_feature(feature_key) + + +def get_feature(feature_key: str) -> Feature | None: + """Get a feature by key.""" + return _get_client().get_feature(feature_key) + + +def get_features() -> list[Feature]: + """Get all features.""" + return _get_client().get_features() + + +def get_license() -> License | None: + """Get the current license.""" + return _get_client().license + + +def get_status() -> LicenseStatus: + """Get the current license status.""" + return _get_client().status + + +def is_licensed() -> bool: + """Check if the license is valid.""" + return _get_client().is_licensed + + +def is_trial() -> bool: + """Check if the license is a trial.""" + return _get_client().is_trial + + +def get_tiers() -> list[ProductTier]: + """Get product tiers (sync).""" + return _get_client().get_tiers() + + +async def get_tiers_async() -> list[ProductTier]: + """Get product tiers (async).""" + return await _get_client().get_tiers_async() + + +def start_purchase(tier_id: str, email: str) -> CheckoutResult: + """Start a purchase checkout (sync).""" + return _get_client().start_purchase(tier_id, email) + + +async def start_purchase_async(tier_id: str, email: str) -> CheckoutResult: + """Start a purchase checkout (async).""" + return await _get_client().start_purchase_async(tier_id, email) diff --git a/src/ironlicensing/cache.py b/src/ironlicensing/cache.py new file mode 100644 index 0000000..f9d5e50 --- /dev/null +++ b/src/ironlicensing/cache.py @@ -0,0 +1,160 @@ +"""License caching for IronLicensing SDK.""" + +import json +from datetime import datetime +from pathlib import Path +from typing import Any + +from .types import License, LicenseResult, Feature + + +class LicenseCache: + """Cache for license validation results.""" + + def __init__(self, key_prefix: str = "ironlicensing", debug: bool = False): + self._key_prefix = key_prefix + self._debug = debug + self._storage_path = Path.home() / ".ironlicensing" + + def _key(self, name: str) -> str: + return f"{self._key_prefix}_{name}" + + def _file_path(self, name: str) -> Path: + return self._storage_path / f"{self._key(name)}.json" + + def _read(self, name: str) -> dict[str, Any] | None: + try: + path = self._file_path(name) + if path.exists(): + return json.loads(path.read_text()) + except Exception: + pass + return None + + def _write(self, name: str, data: dict[str, Any]) -> None: + try: + self._storage_path.mkdir(parents=True, exist_ok=True) + self._file_path(name).write_text(json.dumps(data)) + except Exception: + pass + + def _delete(self, name: str) -> None: + try: + path = self._file_path(name) + if path.exists(): + path.unlink() + except Exception: + pass + + def _serialize_license(self, license: License) -> dict[str, Any]: + return { + "id": license.id, + "key": license.key, + "status": license.status, + "type": license.type, + "features": [ + { + "key": f.key, + "name": f.name, + "enabled": f.enabled, + "description": f.description, + "metadata": f.metadata, + } + for f in license.features + ], + "max_activations": license.max_activations, + "current_activations": license.current_activations, + "created_at": license.created_at.isoformat(), + "email": license.email, + "name": license.name, + "company": license.company, + "expires_at": license.expires_at.isoformat() if license.expires_at else None, + "last_validated_at": license.last_validated_at.isoformat() if license.last_validated_at else None, + "metadata": license.metadata, + } + + def _deserialize_license(self, data: dict[str, Any]) -> License: + features = [ + Feature( + key=f["key"], + name=f["name"], + enabled=f["enabled"], + description=f.get("description"), + metadata=f.get("metadata"), + ) + for f in data.get("features", []) + ] + + return License( + id=data["id"], + key=data["key"], + status=data["status"], + type=data["type"], + features=features, + max_activations=data["max_activations"], + current_activations=data["current_activations"], + created_at=datetime.fromisoformat(data["created_at"]), + email=data.get("email"), + name=data.get("name"), + company=data.get("company"), + expires_at=datetime.fromisoformat(data["expires_at"]) if data.get("expires_at") else None, + last_validated_at=datetime.fromisoformat(data["last_validated_at"]) + if data.get("last_validated_at") + else None, + metadata=data.get("metadata"), + ) + + def get_license(self) -> License | None: + """Get the cached license.""" + data = self._read("license") + if data: + return self._deserialize_license(data) + return None + + def set_license(self, license: License) -> None: + """Set the cached license.""" + self._write("license", self._serialize_license(license)) + + def get_validation_result(self) -> LicenseResult | None: + """Get the cached validation result.""" + data = self._read("validation") + if data and data.get("license"): + return LicenseResult( + valid=data["valid"], + license=self._deserialize_license(data["license"]), + cached=True, + ) + return None + + def set_validation_result(self, result: LicenseResult, timestamp: datetime) -> None: + """Set the cached validation result.""" + data: dict[str, Any] = {"valid": result.valid} + if result.license: + data["license"] = self._serialize_license(result.license) + self._write("validation", data) + self._write("validation_time", {"timestamp": timestamp.isoformat()}) + + def get_validation_time(self) -> datetime | None: + """Get the validation timestamp.""" + data = self._read("validation_time") + if data and data.get("timestamp"): + return datetime.fromisoformat(data["timestamp"]) + return None + + def get_license_key(self) -> str | None: + """Get the stored license key.""" + data = self._read("license_key") + if data: + return data.get("key") + return None + + def set_license_key(self, key: str) -> None: + """Set the license key.""" + self._write("license_key", {"key": key}) + + def clear(self) -> None: + """Clear all cached data.""" + self._delete("license") + self._delete("validation") + self._delete("validation_time") + self._delete("license_key") diff --git a/src/ironlicensing/client.py b/src/ironlicensing/client.py new file mode 100644 index 0000000..20c144b --- /dev/null +++ b/src/ironlicensing/client.py @@ -0,0 +1,288 @@ +"""Main client for IronLicensing SDK.""" + +from datetime import datetime, timedelta +from typing import Callable + +from .types import ( + LicenseOptions, + License, + LicenseResult, + LicenseStatus, + CheckoutResult, + ProductTier, + Feature, +) +from .transport import Transport +from .cache import LicenseCache + + +class LicenseRequiredError(Exception): + """Raised when a license or feature is required but not available.""" + + pass + + +class LicenseClient: + """IronLicensing client for license validation and activation.""" + + def __init__(self, options: LicenseOptions): + self._options = options + self._transport = Transport( + options.api_base_url, + options.public_key, + options.product_slug, + options.debug, + ) + self._cache = LicenseCache(options.storage_key_prefix, options.debug) + self._current_license: License | None = None + self._on_license_change_callbacks: list[Callable[[License | None], None]] = [] + + # Load cached license + self._current_license = self._cache.get_license() + + if options.debug: + print("[IronLicensing] Client initialized") + + def validate(self, license_key: str | None = None) -> LicenseResult: + """Validate a license key (sync).""" + key = license_key or self._cache.get_license_key() + if not key: + return LicenseResult(valid=False, error="No license key provided") + + # Check cache first + if self._options.enable_offline_cache: + cached = self._get_cached_validation() + if cached: + return LicenseResult( + valid=cached.valid, + license=cached.license, + cached=True, + ) + + # Validate online + result = self._transport.validate(key) + + if result.valid and result.license: + self._set_license(result.license, key) + self._cache.set_validation_result(result, datetime.now()) + + return result + + async def validate_async(self, license_key: str | None = None) -> LicenseResult: + """Validate a license key (async).""" + key = license_key or self._cache.get_license_key() + if not key: + return LicenseResult(valid=False, error="No license key provided") + + # Check cache first + if self._options.enable_offline_cache: + cached = self._get_cached_validation() + if cached: + return LicenseResult( + valid=cached.valid, + license=cached.license, + cached=True, + ) + + # Validate online + result = await self._transport.validate_async(key) + + if result.valid and result.license: + self._set_license(result.license, key) + self._cache.set_validation_result(result, datetime.now()) + + return result + + def activate(self, license_key: str, machine_name: str | None = None) -> LicenseResult: + """Activate a license (sync).""" + result = self._transport.activate(license_key, machine_name) + + if result.valid and result.license: + self._set_license(result.license, license_key) + self._cache.set_validation_result(result, datetime.now()) + + return result + + async def activate_async(self, license_key: str, machine_name: str | None = None) -> LicenseResult: + """Activate a license (async).""" + result = await self._transport.activate_async(license_key, machine_name) + + if result.valid and result.license: + self._set_license(result.license, license_key) + self._cache.set_validation_result(result, datetime.now()) + + return result + + def deactivate(self) -> bool: + """Deactivate the current license (sync).""" + key = self._cache.get_license_key() + if not key: + return False + + success = self._transport.deactivate(key) + if success: + self._set_license(None, None) + self._cache.clear() + + return success + + async def deactivate_async(self) -> bool: + """Deactivate the current license (async).""" + key = self._cache.get_license_key() + if not key: + return False + + success = await self._transport.deactivate_async(key) + if success: + self._set_license(None, None) + self._cache.clear() + + return success + + def start_trial(self, email: str) -> LicenseResult: + """Start a trial (sync).""" + result = self._transport.start_trial(email) + + if result.valid and result.license: + self._set_license(result.license, result.license.key) + self._cache.set_validation_result(result, datetime.now()) + + return result + + async def start_trial_async(self, email: str) -> LicenseResult: + """Start a trial (async).""" + result = await self._transport.start_trial_async(email) + + if result.valid and result.license: + self._set_license(result.license, result.license.key) + self._cache.set_validation_result(result, datetime.now()) + + return result + + def has_feature(self, feature_key: str) -> bool: + """Check if a feature is enabled.""" + if not self._current_license: + return False + for feature in self._current_license.features: + if feature.key == feature_key and feature.enabled: + return True + return False + + def require_feature(self, feature_key: str) -> None: + """Require a feature, raises LicenseRequiredError if not available.""" + if not self.has_feature(feature_key): + raise LicenseRequiredError(f"Feature '{feature_key}' requires a valid license") + + def get_feature(self, feature_key: str) -> Feature | None: + """Get a feature by key.""" + if not self._current_license: + return None + for feature in self._current_license.features: + if feature.key == feature_key: + return feature + return None + + def get_features(self) -> list[Feature]: + """Get all features.""" + if not self._current_license: + return [] + return self._current_license.features + + @property + def license(self) -> License | None: + """Get the current license.""" + return self._current_license + + @property + def status(self) -> LicenseStatus: + """Get the current license status.""" + if self._current_license: + return self._current_license.status + return "not_activated" + + @property + def is_licensed(self) -> bool: + """Check if the license is valid.""" + return self._current_license is not None and self._current_license.status in ("valid", "trial") + + @property + def is_trial(self) -> bool: + """Check if the license is a trial.""" + return ( + self._current_license is not None + and (self._current_license.status == "trial" or self._current_license.type == "trial") + ) + + @property + def expires_at(self) -> datetime | None: + """Get the license expiration date.""" + if self._current_license: + return self._current_license.expires_at + return None + + def get_tiers(self) -> list[ProductTier]: + """Get product tiers (sync).""" + return self._transport.get_tiers() + + async def get_tiers_async(self) -> list[ProductTier]: + """Get product tiers (async).""" + return await self._transport.get_tiers_async() + + def start_purchase(self, tier_id: str, email: str) -> CheckoutResult: + """Start a purchase checkout (sync).""" + return self._transport.start_checkout(tier_id, email) + + async def start_purchase_async(self, tier_id: str, email: str) -> CheckoutResult: + """Start a purchase checkout (async).""" + return await self._transport.start_checkout_async(tier_id, email) + + def on_license_change(self, callback: Callable[[License | None], None]) -> Callable[[], None]: + """Register a callback for license changes.""" + self._on_license_change_callbacks.append(callback) + + def unsubscribe() -> None: + if callback in self._on_license_change_callbacks: + self._on_license_change_callbacks.remove(callback) + + return unsubscribe + + def _set_license(self, license: License | None, key: str | None) -> None: + self._current_license = license + + if license: + self._cache.set_license(license) + if key: + self._cache.set_license_key(key) + + for callback in self._on_license_change_callbacks: + try: + callback(license) + except Exception as e: + if self._options.debug: + print(f"[IronLicensing] Error in license change callback: {e}") + + def _get_cached_validation(self) -> LicenseResult | None: + cached = self._cache.get_validation_result() + time = self._cache.get_validation_time() + + if not cached or not time: + return None + + cache_age = datetime.now() - time + max_age = timedelta(minutes=self._options.cache_validation_minutes) + + if cache_age <= max_age: + return cached + + # Check offline grace period + grace_period = timedelta(days=self._options.offline_grace_days) + if cache_age <= grace_period and cached.valid: + if self._options.debug: + print("[IronLicensing] Using offline grace period") + return cached + + return None + + def close(self) -> None: + """Close the client.""" + self._transport.close() diff --git a/src/ironlicensing/transport.py b/src/ironlicensing/transport.py new file mode 100644 index 0000000..72e1281 --- /dev/null +++ b/src/ironlicensing/transport.py @@ -0,0 +1,440 @@ +"""HTTP transport for IronLicensing SDK.""" + +import platform +import uuid +from datetime import datetime +from pathlib import Path + +import httpx + +from .types import ( + License, + LicenseResult, + CheckoutResult, + ProductTier, + Activation, + Feature, +) + + +class Transport: + """HTTP transport for communicating with the IronLicensing API.""" + + def __init__( + self, + base_url: str, + public_key: str, + product_slug: str, + debug: bool, + ): + self._base_url = base_url + self._public_key = public_key + self._product_slug = product_slug + self._debug = debug + self._sync_client = httpx.Client(timeout=30.0) + self._async_client: httpx.AsyncClient | None = None + self._machine_id = self._get_machine_id() + + def _log(self, message: str) -> None: + if self._debug: + print(f"[IronLicensing] {message}") + + def _get_headers(self) -> dict[str, str]: + return { + "Content-Type": "application/json", + "X-Public-Key": self._public_key, + "X-Product-Slug": self._product_slug, + } + + def _get_machine_id(self) -> str: + """Get or generate a persistent machine ID.""" + storage_path = Path.home() / ".ironlicensing" / "machine_id" + try: + if storage_path.exists(): + return storage_path.read_text().strip() + except Exception: + pass + + machine_id = str(uuid.uuid4()) + try: + storage_path.parent.mkdir(parents=True, exist_ok=True) + storage_path.write_text(machine_id) + except Exception: + pass + + return machine_id + + def _parse_license(self, data: dict) -> License: + features = [ + Feature( + key=f["key"], + name=f["name"], + enabled=f.get("enabled", True), + description=f.get("description"), + metadata=f.get("metadata"), + ) + for f in data.get("features", []) + ] + + return License( + id=data["id"], + key=data["key"], + status=data["status"], + type=data["type"], + features=features, + max_activations=data.get("maxActivations", 1), + current_activations=data.get("currentActivations", 0), + created_at=datetime.fromisoformat(data["createdAt"].replace("Z", "+00:00")), + email=data.get("email"), + name=data.get("name"), + company=data.get("company"), + expires_at=datetime.fromisoformat(data["expiresAt"].replace("Z", "+00:00")) + if data.get("expiresAt") + else None, + last_validated_at=datetime.fromisoformat(data["lastValidatedAt"].replace("Z", "+00:00")) + if data.get("lastValidatedAt") + else None, + metadata=data.get("metadata"), + ) + + def _parse_activation(self, data: dict) -> Activation: + return Activation( + id=data["id"], + machine_id=data["machineId"], + activated_at=datetime.fromisoformat(data["activatedAt"].replace("Z", "+00:00")), + last_seen_at=datetime.fromisoformat(data["lastSeenAt"].replace("Z", "+00:00")), + machine_name=data.get("machineName"), + platform=data.get("platform"), + ) + + def validate(self, license_key: str) -> LicenseResult: + """Validate a license key (sync).""" + self._log(f"Validating license: {license_key[:10]}...") + + try: + response = self._sync_client.post( + f"{self._base_url}/api/v1/validate", + headers=self._get_headers(), + json={ + "licenseKey": license_key, + "machineId": self._machine_id, + }, + ) + + if response.status_code == 200: + data = response.json() + return LicenseResult( + valid=data.get("valid", False), + license=self._parse_license(data["license"]) if data.get("license") else None, + activations=[self._parse_activation(a) for a in data.get("activations", [])], + ) + else: + error_data = response.json() + return LicenseResult(valid=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return LicenseResult(valid=False, error=str(e)) + + async def validate_async(self, license_key: str) -> LicenseResult: + """Validate a license key (async).""" + self._log(f"Validating license: {license_key[:10]}...") + + if self._async_client is None: + self._async_client = httpx.AsyncClient(timeout=30.0) + + try: + response = await self._async_client.post( + f"{self._base_url}/api/v1/validate", + headers=self._get_headers(), + json={ + "licenseKey": license_key, + "machineId": self._machine_id, + }, + ) + + if response.status_code == 200: + data = response.json() + return LicenseResult( + valid=data.get("valid", False), + license=self._parse_license(data["license"]) if data.get("license") else None, + activations=[self._parse_activation(a) for a in data.get("activations", [])], + ) + else: + error_data = response.json() + return LicenseResult(valid=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return LicenseResult(valid=False, error=str(e)) + + def activate(self, license_key: str, machine_name: str | None = None) -> LicenseResult: + """Activate a license (sync).""" + self._log(f"Activating license: {license_key[:10]}...") + + try: + response = self._sync_client.post( + f"{self._base_url}/api/v1/activate", + headers=self._get_headers(), + json={ + "licenseKey": license_key, + "machineId": self._machine_id, + "machineName": machine_name or platform.node(), + "platform": platform.system(), + }, + ) + + if response.status_code == 200: + data = response.json() + return LicenseResult( + valid=data.get("valid", False), + license=self._parse_license(data["license"]) if data.get("license") else None, + activations=[self._parse_activation(a) for a in data.get("activations", [])], + ) + else: + error_data = response.json() + return LicenseResult(valid=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return LicenseResult(valid=False, error=str(e)) + + async def activate_async(self, license_key: str, machine_name: str | None = None) -> LicenseResult: + """Activate a license (async).""" + self._log(f"Activating license: {license_key[:10]}...") + + if self._async_client is None: + self._async_client = httpx.AsyncClient(timeout=30.0) + + try: + response = await self._async_client.post( + f"{self._base_url}/api/v1/activate", + headers=self._get_headers(), + json={ + "licenseKey": license_key, + "machineId": self._machine_id, + "machineName": machine_name or platform.node(), + "platform": platform.system(), + }, + ) + + if response.status_code == 200: + data = response.json() + return LicenseResult( + valid=data.get("valid", False), + license=self._parse_license(data["license"]) if data.get("license") else None, + activations=[self._parse_activation(a) for a in data.get("activations", [])], + ) + else: + error_data = response.json() + return LicenseResult(valid=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return LicenseResult(valid=False, error=str(e)) + + def deactivate(self, license_key: str) -> bool: + """Deactivate a license (sync).""" + try: + response = self._sync_client.post( + f"{self._base_url}/api/v1/deactivate", + headers=self._get_headers(), + json={ + "licenseKey": license_key, + "machineId": self._machine_id, + }, + ) + return response.status_code == 200 + except Exception: + return False + + async def deactivate_async(self, license_key: str) -> bool: + """Deactivate a license (async).""" + if self._async_client is None: + self._async_client = httpx.AsyncClient(timeout=30.0) + + try: + response = await self._async_client.post( + f"{self._base_url}/api/v1/deactivate", + headers=self._get_headers(), + json={ + "licenseKey": license_key, + "machineId": self._machine_id, + }, + ) + return response.status_code == 200 + except Exception: + return False + + def start_trial(self, email: str) -> LicenseResult: + """Start a trial (sync).""" + self._log(f"Starting trial for: {email}") + + try: + response = self._sync_client.post( + f"{self._base_url}/api/v1/trial", + headers=self._get_headers(), + json={ + "email": email, + "machineId": self._machine_id, + }, + ) + + if response.status_code == 200: + data = response.json() + return LicenseResult( + valid=data.get("valid", False), + license=self._parse_license(data["license"]) if data.get("license") else None, + ) + else: + error_data = response.json() + return LicenseResult(valid=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return LicenseResult(valid=False, error=str(e)) + + async def start_trial_async(self, email: str) -> LicenseResult: + """Start a trial (async).""" + self._log(f"Starting trial for: {email}") + + if self._async_client is None: + self._async_client = httpx.AsyncClient(timeout=30.0) + + try: + response = await self._async_client.post( + f"{self._base_url}/api/v1/trial", + headers=self._get_headers(), + json={ + "email": email, + "machineId": self._machine_id, + }, + ) + + if response.status_code == 200: + data = response.json() + return LicenseResult( + valid=data.get("valid", False), + license=self._parse_license(data["license"]) if data.get("license") else None, + ) + else: + error_data = response.json() + return LicenseResult(valid=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return LicenseResult(valid=False, error=str(e)) + + def _parse_tier(self, data: dict) -> ProductTier: + features = [ + Feature( + key=f["key"], + name=f["name"], + enabled=f.get("enabled", True), + description=f.get("description"), + metadata=f.get("metadata"), + ) + for f in data.get("features", []) + ] + + return ProductTier( + id=data["id"], + slug=data["slug"], + name=data["name"], + price=data["price"], + currency=data["currency"], + features=features, + description=data.get("description"), + billing_period=data.get("billingPeriod"), + ) + + def get_tiers(self) -> list[ProductTier]: + """Get product tiers (sync).""" + try: + response = self._sync_client.get( + f"{self._base_url}/api/v1/tiers", + headers=self._get_headers(), + ) + + if response.status_code == 200: + data = response.json() + return [self._parse_tier(t) for t in data.get("tiers", [])] + return [] + + except Exception: + return [] + + async def get_tiers_async(self) -> list[ProductTier]: + """Get product tiers (async).""" + if self._async_client is None: + self._async_client = httpx.AsyncClient(timeout=30.0) + + try: + response = await self._async_client.get( + f"{self._base_url}/api/v1/tiers", + headers=self._get_headers(), + ) + + if response.status_code == 200: + data = response.json() + return [self._parse_tier(t) for t in data.get("tiers", [])] + return [] + + except Exception: + return [] + + def start_checkout(self, tier_id: str, email: str) -> CheckoutResult: + """Start a checkout session (sync).""" + try: + response = self._sync_client.post( + f"{self._base_url}/api/v1/checkout", + headers=self._get_headers(), + json={ + "tierId": tier_id, + "email": email, + }, + ) + + if response.status_code == 200: + data = response.json() + return CheckoutResult( + success=True, + checkout_url=data.get("checkoutUrl"), + session_id=data.get("sessionId"), + ) + else: + error_data = response.json() + return CheckoutResult(success=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return CheckoutResult(success=False, error=str(e)) + + async def start_checkout_async(self, tier_id: str, email: str) -> CheckoutResult: + """Start a checkout session (async).""" + if self._async_client is None: + self._async_client = httpx.AsyncClient(timeout=30.0) + + try: + response = await self._async_client.post( + f"{self._base_url}/api/v1/checkout", + headers=self._get_headers(), + json={ + "tierId": tier_id, + "email": email, + }, + ) + + if response.status_code == 200: + data = response.json() + return CheckoutResult( + success=True, + checkout_url=data.get("checkoutUrl"), + session_id=data.get("sessionId"), + ) + else: + error_data = response.json() + return CheckoutResult(success=False, error=error_data.get("error", f"HTTP {response.status_code}")) + + except Exception as e: + return CheckoutResult(success=False, error=str(e)) + + def close(self) -> None: + """Close the transport.""" + self._sync_client.close() + if self._async_client: + # Note: async client should be closed in an async context + pass diff --git a/src/ironlicensing/types.py b/src/ironlicensing/types.py new file mode 100644 index 0000000..2d19674 --- /dev/null +++ b/src/ironlicensing/types.py @@ -0,0 +1,113 @@ +"""Type definitions for IronLicensing SDK.""" + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Literal + +LicenseStatus = Literal[ + "valid", + "expired", + "suspended", + "revoked", + "invalid", + "trial", + "trial_expired", + "not_activated", + "unknown", +] + +LicenseType = Literal["perpetual", "subscription", "trial"] + +BillingPeriod = Literal["monthly", "yearly", "lifetime"] + + +@dataclass +class LicenseOptions: + """Configuration options for the LicenseClient.""" + + public_key: str + product_slug: str + api_base_url: str = "https://api.ironlicensing.com" + debug: bool = False + enable_offline_cache: bool = True + cache_validation_minutes: int = 60 + offline_grace_days: int = 7 + storage_key_prefix: str = "ironlicensing" + + +@dataclass +class Feature: + """A feature in a license.""" + + key: str + name: str + enabled: bool + description: str | None = None + metadata: dict[str, Any] | None = None + + +@dataclass +class License: + """License information.""" + + id: str + key: str + status: LicenseStatus + type: LicenseType + features: list[Feature] + max_activations: int + current_activations: int + created_at: datetime + email: str | None = None + name: str | None = None + company: str | None = None + expires_at: datetime | None = None + last_validated_at: datetime | None = None + metadata: dict[str, Any] | None = None + + +@dataclass +class Activation: + """Activation information.""" + + id: str + machine_id: str + activated_at: datetime + last_seen_at: datetime + machine_name: str | None = None + platform: str | None = None + + +@dataclass +class LicenseResult: + """Result of license validation.""" + + valid: bool + license: License | None = None + activations: list[Activation] | None = None + error: str | None = None + cached: bool = False + + +@dataclass +class CheckoutResult: + """Result of checkout.""" + + success: bool + checkout_url: str | None = None + session_id: str | None = None + error: str | None = None + + +@dataclass +class ProductTier: + """Product tier for purchase.""" + + id: str + slug: str + name: str + price: float + currency: str + features: list[Feature] = field(default_factory=list) + description: str | None = None + billing_period: BillingPeriod | None = None