profy-sdk 0.1.0

Profy App SDK — OAuth token management and event reporting

profy-sdk

Official Python SDK for the Profy App platform — OAuth token management and event reporting.

Installation

pip install profy-sdk

# With optional SQLAlchemy token storage:
pip install profy-sdk[sqlalchemy]

Quick Start

from profy import ProfyApp

async with ProfyApp(
    client_id="your-app-id",
    client_secret="your-app-secret",
) as profy:
    # Exchange authorization code for tokens
    token = await profy.exchange_code(code, "https://your-app.com/callback")

    # Report a billing event
    await profy.report_event("generate_report", token=token)

Synchronous Usage

from profy import ProfyAppSync

with ProfyAppSync(
    client_id="your-app-id",
    client_secret="your-app-secret",
) as profy:
    token = profy.exchange_code(code, redirect_uri)
    profy.report_event("generate_report", token=token)

Features

• OAuth2 Authorization Code Flow — exchange codes, refresh tokens automatically
• Events API — report billing events with auto token refresh and retry on 401
• Async + Sync — ProfyApp (async) and ProfyAppSync (sync) clients
• Type hints — full type annotations, PEP 561 compatible
• Minimal dependencies — only httpx
• Optional SQLAlchemy integration — pre-built token model + store

Token Persistence (Optional)

Using SQLAlchemy

from profy import ProfyApp
from profy.contrib.sqlalchemy import create_token_model, SQLAlchemyTokenStore

# 1. Create the ORM model (auto-creates table with Base.metadata.create_all())
ProfyOAuthToken = create_token_model(Base)

# 2. Create store
store = SQLAlchemyTokenStore(async_session_factory, ProfyOAuthToken)

# 3. Initialize SDK with persistence
profy = ProfyApp(
    client_id="...",
    client_secret="...",
    on_token_refresh=store.on_refresh,
)

# 4. Save after exchange
token = await profy.exchange_code(code, redirect_uri)
await store.save(token.user_id, token, scope="events:write")

# 5. Load later
saved = await store.load(user_id)

Custom Storage

from profy import ProfyApp, TokenData

async def persist_token(token: TokenData):
    await redis.set(f"profy:{token.user_id}", token.access_token)

profy = ProfyApp(
    client_id="...",
    client_secret="...",
    on_token_refresh=persist_token,
)

API Reference

ProfyApp / ProfyAppSync

Method	Description
exchange_code(code, redirect_uri)	Exchange authorization code for tokens
refresh_token(refresh_token)	Manually refresh an expired token
report_event(event_name, *, token, idempotency_key?, metadata?)	Report a billing event

TokenData

@dataclass
class TokenData:
    access_token: str
    refresh_token: str
    user_id: str
    expires_at: float  # unix timestamp

    @property
    def is_expired(self) -> bool: ...

Exceptions

Class	Status	Description
AuthExpired	401	Token expired and refresh failed
InsufficientBalance	402	User has insufficient credits
InvalidEvent	400	Event name not configured
ProfyApiError	varies	Generic API error

Documentation

• Integration Quickstart
• SDK Guide
• API Reference

License

MIT