ironservices
/
ironlicensing-python
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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
This commit is contained in:
David Friedel 2025-12-25 11:17:34 +00:00
parent ba3065912a
commit 21cd3f004a
8 changed files with 1554 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
.gitignore vendored Normal file
View File

@ -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/

279
README.md
View File

@ -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.

38
pyproject.toml Normal file
View File

@ -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"]

202
src/ironlicensing/__init__.py Normal file
View File

@ -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)

160
src/ironlicensing/cache.py Normal file
View File

@ -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")

288
src/ironlicensing/client.py Normal file
View File

@ -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()

440
src/ironlicensing/transport.py Normal file
View File

@ -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

113
src/ironlicensing/types.py Normal file
View File

@ -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