pyrx-auth-sdk

SDK for PYRX ecosystem services to integrate with pyrx.auth centralized authentication.

Installation

pip install pyrx-auth-sdk

# With FastAPI dependency support
pip install pyrx-auth-sdk[fastapi]

Features

• JWT Verification: ES256 JWT verification with JWKS caching (no network per request)
• FastAPI Dependency: get_current_user() extracts AuthenticatedUser from Bearer JWT
• Event Consumer: Redis Stream consumer for user lifecycle events
• Internal API Client: Service-to-service user lookup API

Quick Start

FastAPI Integration

from pyrx_auth_sdk.config import AuthSDKConfig
from pyrx_auth_sdk.dependencies import init_auth, get_current_user
from pyrx_auth_sdk.schemas import AuthenticatedUser

# At app startup
config = AuthSDKConfig(
    jwks_url="http://pyrx-auth:8001/api/v1/.well-known/jwks.json",
    issuer="pyrx.auth",
    audience="pyrx-mentor",
)
init_auth(config)

# In route handlers
@app.get("/protected")
async def protected(user: AuthenticatedUser = Depends(get_current_user)):
    print(f"User {user.id} from tenant {user.tenant_id}")

Event Consumer

from pyrx_auth_sdk.events import UserEventConsumer

consumer = UserEventConsumer(
    redis=redis_client,
    group_name="my-service-group",
    consumer_name="worker-1",
)

async def on_user_registered(payload):
    print(f"New user: {payload['user_id']}")

consumer.on("user.registered", on_user_registered)
await consumer.run()

Internal API Client

from pyrx_auth_sdk import AuthInternalClient
from pyrx_auth_sdk.config import AuthSDKConfig

client = AuthInternalClient(AuthSDKConfig(
    internal_api_url="http://pyrx-auth:8001",
    internal_api_key="your-service-key",
))

user = await client.get_user(user_id)
users, not_found = await client.get_users_batch(["uuid-1", "uuid-2"])

Environment Variables