opengater-globalauth-client

Python client library for GlobalAuth authentication service.

Features

• 🔐 HTTP client for token validation and referral operations
• 🐰 RabbitMQ integration for events and commands
• ⚡ FastAPI middleware and dependencies
• 💾 Built-in caching for introspect results
• 🔄 Async-first design

Installation

pip install opengater-globalauth-client

Or with uv:

uv add opengater-globalauth-client

Quick Start

HTTP Client

import os
from opengater_globalauth_client import GlobalAuthClient

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="opengater",
    internal_api_key=os.environ.get("GLOBALAUTH_INTERNAL_API_KEY"),  # since 0.4.0
    cache_ttl=300  # Cache introspect results for 5 minutes
)

# Validate token
user = await client.introspect(token)
print(f"User: {user.id}, verified: {user.verified}")

# Decode referral code
ref_info = await client.decode_referral("ref_xxx")
if ref_info.type == "referral":
    print(f"Invited by: {ref_info.inviter_id}")

# Get referral link for user
link = await client.get_referral_link(user_token)
print(f"Share this link: {link.link}")

# Get referral stats
stats = await client.get_referral_stats(user_token)
print(f"Total invited: {stats.total_invited}")

# Auth method bonuses (inter-service, needs internal_api_key; since 0.5.1)
weights = await client.get_auth_method_weights()   # list[AuthMethodWeight]
profile = await client.get_auth_profile(user_id)   # AuthProfile
print(profile.registration_auth_type, profile.linked_auth_types)

RabbitMQ Integration

from faststream.rabbit import RabbitBroker
from opengater_globalauth_client.broker import (
    GlobalAuthConsumer,
    GlobalAuthPublisher,
    UserCreatedEvent,
    CreateUserCommand,
)

# Use your existing broker
broker = RabbitBroker(settings.rabbitmq_url)

# Consumer for events
consumer = GlobalAuthConsumer(broker, "globalauth.opengater")

@consumer.on_user_created
async def handle_user_created(event: UserCreatedEvent):
    print(f"New user: {event.user_id}")
    
    if event.invited_by_id:
        # Award referral bonus
        await award_bonus(event.invited_by_id)

@consumer.on_auth_method_linked
async def handle_linked(event):
    print(f"User {event.user_id} linked {event.auth_type}")

# Publisher for commands (service_slug is required)
publisher = GlobalAuthPublisher(broker=broker, service_slug="opengater")

# Create user from Telegram bot
cmd = CreateUserCommand(
    auth_type="telegram",
    identifier="123456789",
    language="ru",
    extra_data={"username": "john", "first_name": "John"},
    invited_by_id="uuid-of-inviter",
)
await publisher.create_user(cmd)

FastAPI Integration

With Middleware (Recommended)

from fastapi import FastAPI, Depends, Request
from opengater_globalauth_client import GlobalAuthClient
from opengater_globalauth_client.fastapi import AuthMiddleware, get_current_user

app = FastAPI()

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="opengater"
)

# Add middleware for automatic token validation
app.add_middleware(
    AuthMiddleware,
    client=client,
    exclude_paths=["/health", "/public/*"]  # Optional
)

@app.get("/me")
async def me(user = Depends(get_current_user())):
    return {
        "id": user.id,
        "verified": user.verified,
        "trust_level": user.trust.level
    }

@app.get("/admin-only")
async def admin_only(user = Depends(require_admin())):
    return {"message": "Welcome, admin!"}

Without Middleware

from fastapi import FastAPI, Depends
from opengater_globalauth_client import GlobalAuthClient
from opengater_globalauth_client.fastapi import get_current_user

app = FastAPI()

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="opengater"
)

# Pass client to dependency
@app.get("/me")
async def me(user = Depends(get_current_user(client))):
    return {"id": user.id}

Available Events

User Model

class User:
    id: str
    verified: bool
    is_banned: bool
    role: Literal["user", "support", "admin"]
    trust: TrustInfo  # score, level
    auth_methods: list[AuthMethod]
    invited_by_id: str | None
    registered_via_event: str | None
    registered_via_service: str | None

Configuration

Inter-service API key (since 0.4.0)

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="my-service",
    internal_api_key=os.environ.get("GLOBALAUTH_INTERNAL_API_KEY"),
)

internal_api_key (optional) — inter-service API ключ. Когда задан, отправляется как X-API-Key на /auth/jwt/introspect. Нужен начиная с GlobalAuth E5 (hard requirement). До E5 — игнорируется сервером.

Error Handling

from opengater_globalauth_client import (
    GlobalAuthClient,
    TokenExpiredError,
    TokenInvalidError,
    UserBannedError,
    GlobalAuthConnectionError,
)

try:
    user = await client.introspect(token)
except TokenExpiredError:
    # Token expired, need to refresh
    pass
except UserBannedError:
    # User is banned
    pass
except TokenInvalidError:
    # Invalid token
    pass
except GlobalAuthConnectionError:
    # GlobalAuth service unavailable
    pass

Development

This package uses uv for dependency management.

# Install dev dependencies
uv sync --group dev

# Run tests
uv run pytest

# Lint & type-check
uv run ruff check src tests
uv run mypy src

# Build a release
uv build  # → dist/opengater_globalauth_client-X.Y.Z-*.whl

Publishing to PyPI (maintainer-only):

uv build
twine upload dist/opengater_globalauth_client-X.Y.Z*

License

MIT