Python SDK
The PUGUH Python SDK provides an async-first client for integrating authentication, organizations, applications, and billing into your Python backend services. It supports both user-session and service-to-service authentication patterns.
Installation
pip install puguh-sdk Quick Start
User Context (Frontend / User Sessions)
Use an access token obtained from user login to make requests on behalf of a user:
from puguh_sdk import PuguhClient
async with PuguhClient(
base_url="https://api-puguh.arsaka.io",
access_token="user_jwt_token"
) as client:
# List organizations the user belongs to
orgs = await client.organizations.list()
for org in orgs:
print(f"{org.name} ({org.slug})") Service Context (Backend / Service-to-Service)
Use an API key for backend services that need to call PUGUH without a user session:
from puguh_sdk import PuguhClient
async with PuguhClient(
base_url="https://api-puguh.arsaka.io",
api_key="pk_live_xxxxxxxxxxxx"
) as client:
# Fetch available billing plans
plans = await client.billing.get_plans()
for plan in plans:
print(f"{plan.name}: {plan.price}/mo") Two Authentication Modes
Use access_token when acting on behalf of a logged-in user (e.g., in API routes that receive a JWT from the frontend). Use api_key for backend services, cron jobs, or admin scripts that operate without a user session.
Service Modules
The SDK exposes four service modules, each accessible as a property on the client instance.
Auth (client.auth)
Handle user authentication, token lifecycle, MFA, and magic links.
# Login — returns access_token and refresh_token
tokens = await client.auth.login(
email="user@example.com",
password="secure_password"
)
print(tokens.access_token)
# Register a new user
user = await client.auth.register(
email="new@example.com",
password="secure_password",
full_name="Jane Doe"
)
print(user.user_id)
# Refresh an expired access token
new_tokens = await client.auth.refresh_token(
refresh_token=tokens.refresh_token
)
# Validate a token and get user context
user_ctx = await client.auth.validate_token(token=tokens.access_token)
print(user_ctx.email, user_ctx.roles)
# Magic link authentication
await client.auth.request_magic_link(email="user@example.com")
tokens = await client.auth.verify_magic_link(token="ml_token_from_email")
# Multi-Factor Authentication
setup = await client.auth.setup_mfa()
print(setup.qr_code_url) # Show QR to user
await client.auth.confirm_mfa(code="123456")
await client.auth.disable_mfa() Note
There is no get_me() method. User information is available from the login response or by decoding the JWT claims locally using the PUGUH JWKS endpoint.
Organizations (client.organizations)
Manage organizations and their members.
# List organizations
orgs = await client.organizations.list()
# Create a new organization
org = await client.organizations.create(
name="Acme Corp",
slug="acme-corp"
)
# Get organization details
org = await client.organizations.get(org_id="org-uuid")
# Update organization
await client.organizations.update(
org_id="org-uuid",
name="Acme Corporation"
)
# Delete organization
await client.organizations.delete(org_id="org-uuid")
# Member management
members = await client.organizations.list_members(org_id="org-uuid")
await client.organizations.invite_member(
org_id="org-uuid",
email="colleague@example.com",
role="member"
)
await client.organizations.update_member_role(
org_id="org-uuid",
user_id="user-uuid",
role="admin"
)
await client.organizations.remove_member(
org_id="org-uuid",
user_id="user-uuid"
) Applications (client.applications)
Manage applications within an organization.
# List applications in an organization
apps = await client.applications.list(org_id="org-uuid")
# Create an application
app = await client.applications.create(
org_id="org-uuid",
name="My SaaS App",
slug="my-saas-app"
)
# Get application details
app = await client.applications.get(app_id="app-uuid")
# Update application
await client.applications.update(
app_id="app-uuid",
name="My SaaS App v2"
)
# Delete application
await client.applications.delete(app_id="app-uuid")
# Application member management
members = await client.applications.list_members(app_id="app-uuid")
await client.applications.add_member(
app_id="app-uuid",
user_id="user-uuid",
role="member"
) Billing (client.billing)
Manage subscriptions, plans, invoices, and usage metering.
# List available plans
plans = await client.billing.get_plans()
for plan in plans:
print(f"{plan.name}: {plan.price}")
# Get a specific plan
plan = await client.billing.get_plan(plan_id="plan-uuid")
# Get current subscription for an organization
sub = await client.billing.get_subscription(org_id="org-uuid")
print(f"Plan: {sub.plan_name}, Status: {sub.status}")
# Create a new subscription
sub = await client.billing.create_subscription(
org_id="org-uuid",
plan_id="plan-uuid"
)
# Upgrade to a different plan
await client.billing.upgrade(
org_id="org-uuid",
plan_id="new-plan-uuid"
)
# Cancel subscription
await client.billing.cancel(org_id="org-uuid")
# Invoices and usage
invoices = await client.billing.get_invoices(org_id="org-uuid")
usage = await client.billing.get_usage(org_id="org-uuid") Context Management
Set organization and application context to avoid passing IDs on every call. Once set, the context is sent automatically with all subsequent requests.
from puguh_sdk import PuguhClient
async with PuguhClient(
base_url="https://api-puguh.arsaka.io",
api_key="pk_live_xxxxxxxxxxxx"
) as client:
# Set organization context
client.set_organization_context("org-uuid")
# Set application context
client.set_application_context("app-uuid")
# Now calls automatically include these IDs
members = await client.organizations.list_members()
# Clear context when switching
client.clear_context() Error Handling
The SDK raises typed exceptions for different error scenarios. Always wrap SDK calls in try/except blocks for production code.
from puguh_sdk import PuguhClient
from puguh_sdk.exceptions import (
PuguhError,
AuthError,
RateLimitError,
NetworkError,
)
async with PuguhClient(
base_url="https://api-puguh.arsaka.io",
access_token="user_jwt_token"
) as client:
try:
orgs = await client.organizations.list()
except AuthError as e:
# Token expired or invalid — redirect to login
print(f"Authentication failed: {e}")
except RateLimitError as e:
# Too many requests — back off and retry
print(f"Rate limited. Retry after {e.retry_after}s")
except NetworkError as e:
# Connection failed — check network or API status
print(f"Network error: {e}")
except PuguhError as e:
# Catch-all for any other PUGUH API error
print(f"API error {e.status_code}: {e.message}") Exception Hierarchy
| Exception | When | Typical Action |
|---|---|---|
AuthError | 401 Unauthorized or invalid token | Refresh token or redirect to login |
RateLimitError | 429 Too Many Requests | Wait retry_after seconds, then retry |
NetworkError | Connection timeout or DNS failure | Retry with backoff or alert ops |
PuguhError | Any other 4xx/5xx response | Log and handle based on status code |
Environment Variables
The recommended way to configure the SDK in production:
# .env
PUGUH_BASE_URL=https://api-puguh.arsaka.io
PUGUH_API_KEY=pk_live_xxxxxxxxxxxx import os
from puguh_sdk import PuguhClient
async with PuguhClient(
base_url=os.environ["PUGUH_BASE_URL"],
api_key=os.environ["PUGUH_API_KEY"],
) as client:
plans = await client.billing.get_plans() Best Practices
1. Use Environment Variables for Credentials
Never hardcode API keys or tokens in source code:
# Bad — credentials in source code
client = PuguhClient(api_key="pk_live_abc123")
# Good — credentials from environment
client = PuguhClient(api_key=os.environ["PUGUH_API_KEY"]) 2. Always Use the Async Context Manager
The context manager ensures HTTP connections are properly closed:
# Bad — connection may leak
client = PuguhClient(base_url="...", api_key="...")
orgs = await client.organizations.list()
# forgot to close
# Good — auto-cleanup on exit
async with PuguhClient(base_url="...", api_key="...") as client:
orgs = await client.organizations.list()
# connection closed automatically 3. Handle Errors at the Appropriate Level
Catch specific exceptions close to the call site, and let unexpected errors bubble up:
async def get_user_orgs(access_token: str) -> list:
async with PuguhClient(
base_url=os.environ["PUGUH_BASE_URL"],
access_token=access_token,
) as client:
try:
return await client.organizations.list()
except AuthError:
# Expected: token expired
raise HTTPException(status_code=401, detail="Session expired")
# Let NetworkError, PuguhError bubble up to global handler 4. Reuse Client Instances
For long-running services, create the client once and reuse it across requests:
# In your FastAPI app
from contextlib import asynccontextmanager
from puguh_sdk import PuguhClient
puguh_client: PuguhClient | None = None
@asynccontextmanager
async def lifespan(app):
global puguh_client
puguh_client = PuguhClient(
base_url=os.environ["PUGUH_BASE_URL"],
api_key=os.environ["PUGUH_API_KEY"],
)
await puguh_client.__aenter__()
yield
await puguh_client.__aexit__(None, None, None)
app = FastAPI(lifespan=lifespan)
@app.get("/plans")
async def list_plans():
return await puguh_client.billing.get_plans() Python Version Requirement
The PUGUH Python SDK requires Python 3.11+ and uses asyncio natively. All methods are async and must be awaited.