opensora-uaa 1.0.1

Universal Authentication & Authorization SDK for Python

opensora-uaa

Universal Authentication & Authorization SDK for Python/FastAPI.

Installation

pip install opensora-uaa

Usage

Basic Client Usage

from opensora_uaa import UAAClient

# Initialize client
uaa = UAAClient(project_id="your-project-id")

# Login with Google
async def login():
    result = await uaa.login_with_google(credential="google_jwt_token")
    print(f"User: {result.user.email}")
    print(f"Access Token: {result.access_token}")

# Verify token
async def verify():
    verification = await uaa.verify_token("access_token_here")
    if verification.valid:
        print(f"User: {verification.user.email}")
    else:
        print("Invalid token")

# Get current user
async def get_user():
    user = await uaa.get_current_user("access_token_here")
    print(f"User: {user.email}")

# Refresh token
async def refresh():
    result = await uaa.refresh_token("refresh_token_here")
    print(f"New access token: {result.access_token}")

# Logout
async def logout():
    await uaa.logout("access_token_here")
    print("Logged out successfully")

FastAPI Integration

Method 1: Using UAAAuth Dependency

from fastapi import FastAPI, Depends
from opensora_uaa import UAAAuth, User

app = FastAPI()
auth = UAAAuth()

@app.get("/api/profile")
async def get_profile(user: User = Depends(auth.get_current_user)):
    """Protected endpoint - requires authentication"""
    return {
        "email": user.email,
        "name": user.name,
        "provider": user.provider
    }

@app.get("/api/admin")
async def admin_only(user: User = Depends(auth.get_current_user)):
    """Check if user is admin (implement your own logic)"""
    if user.email not in ["[e-mail]"]:
        raise HTTPException(403, "Admin access required")

    return {"message": "Welcome admin"}

Method 2: Manual Token Verification

from fastapi import FastAPI, Header, HTTPException
from opensora_uaa import UAAClient

app = FastAPI()
uaa = UAAClient(project_id="your-project-id")

@app.get("/api/profile")
async def get_profile(authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(401, "Missing authorization header")

    token = authorization.replace("Bearer ", "")
    verification = await uaa.verify_token(token)

    if not verification.valid:
        raise HTTPException(401, "Invalid or expired token")

    return {"user": verification.user}

Complete FastAPI Example

from fastapi import FastAPI, Depends, HTTPException
from opensora_uaa import UAAClient, UAAAuth, User
from pydantic import BaseModel

app = FastAPI()
uaa = UAAClient(project_id="your-project-id")
auth = UAAAuth()

# Request models
class GoogleLoginRequest(BaseModel):
    credential: str

class AppleLoginRequest(BaseModel):
    code: str
    id_token: str

# Public endpoints
@app.post("/auth/google")
async def google_login(request: GoogleLoginRequest):
    """Login with Google"""
    try:
        result = await uaa.login_with_google(request.credential)
        return {
            "access_token": result.access_token,
            "refresh_token": result.refresh_token,
            "user": result.user
        }
    except Exception as e:
        raise HTTPException(400, str(e))

@app.post("/auth/apple")
async def apple_login(request: AppleLoginRequest):
    """Login with Apple"""
    try:
        result = await uaa.login_with_apple(request.code, request.id_token)
        return {
            "access_token": result.access_token,
            "refresh_token": result.refresh_token,
            "user": result.user
        }
    except Exception as e:
        raise HTTPException(400, str(e))

# Protected endpoints
@app.get("/api/me")
async def get_current_user_info(user: User = Depends(auth.get_current_user)):
    """Get current authenticated user"""
    return {"user": user}

@app.post("/auth/logout")
async def logout(
    authorization: str = Header(None),
    user: User = Depends(auth.get_current_user)
):
    """Logout current user"""
    token = authorization.replace("Bearer ", "")
    await uaa.logout(token)
    return {"message": "Logged out successfully"}

# Startup/Shutdown events
@app.on_event("shutdown")
async def shutdown():
    await uaa.close()

Context Manager Usage

async with UAAClient(project_id="your-project-id") as uaa:
    result = await uaa.login_with_google("credential")
    print(result.user.email)
# Client is automatically closed

API Reference

UAAClient

Main client for UAA operations.

Constructor

UAAClient(
    project_id: str,
    base_url: str = "https://auth.opensora.store/api/v1"
)

Methods

• async login_with_google(credential: str) -> AuthResponse
• async login_with_apple(code: str, id_token: str) -> AuthResponse
• async verify_token(token: str) -> VerifyResponse
• async get_current_user(token: str) -> User
• async refresh_token(refresh_token: str) -> AuthResponse
• async logout(token: str) -> None
• async close() -> None

UAAAuth

FastAPI dependency for authentication.

Constructor

UAAAuth(base_url: str = "https://auth.opensora.store/api/v1")

Methods

• async get_current_user(authorization: Optional[str] = None) -> User
  • Use as FastAPI dependency: user: User = Depends(auth.get_current_user)

Models

User

class User(BaseModel):
    id: str
    email: str
    name: Optional[str]
    avatar: Optional[str]
    provider: Literal['google', 'apple', 'wechat', 'github']
    created_at: datetime

AuthResponse

class AuthResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str
    expires_in: int
    user: User

VerifyResponse

class VerifyResponse(BaseModel):
    valid: bool
    user: Optional[User]

Requirements

• Python >= 3.8
• httpx >= 0.24.0
• pydantic >= 2.0.0

License

MIT