identity-plan-kit 0.1.6

Modern FastAPI library for authentication, RBAC, subscription plans, and usage tracking

IdentityPlanKit

Modern FastAPI library for authentication, RBAC, subscription plans, and usage tracking.

Features

• OAuth Authentication: Google OAuth (extensible for other providers)
• Role-Based Access Control (RBAC): Fine-grained permissions system
• Subscription Plans: Plan management with feature limits and quotas
• Usage Tracking: Track feature usage against plan limits
• Production-Ready: Health checks, graceful shutdown, connection retry
• Audit Logging: Security event logging
• Account Lockout: Brute-force protection
• CLI Tools: Database migrations and management

Installation

pip install identity-plan-kit

Or with uv:

uv pip install identity-plan-kit

Quick Start

1. Set Environment Variables

Create a .env file:

IPK_DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/mydb
IPK_SECRET_KEY=your-super-secret-key-at-least-32-characters-long
IPK_GOOGLE_CLIENT_ID=your-google-client-id
IPK_GOOGLE_CLIENT_SECRET=your-google-client-secret
IPK_GOOGLE_REDIRECT_URI=http://localhost:8000/auth/google/callback
IPK_ENVIRONMENT=development

2. Run Database Migrations

The library includes a CLI tool for managing database migrations:

# Check current migration status
ipk db current

# Apply all migrations
ipk db upgrade

# Use a specific .env file
ipk db upgrade --env-file .env.production

# View migration history
ipk db history

# Show configuration
ipk info

All CLI commands support the --env-file (or -e) option to load environment variables from a specific file. If not specified, the CLI will automatically load from .env if it exists in the current directory.

CLI Commands Reference

Command	Description
ipk db upgrade [revision]	Upgrade database to a later version (default: head)
ipk db downgrade [revision]	Downgrade database to a previous version
ipk db current	Show current database revision
ipk db history	Show migration history
ipk db heads	Show current available heads
ipk db stamp <revision>	Mark database as being at a specific version (without running migrations)
ipk version	Show IdentityPlanKit version
ipk info	Show configuration and environment information

Migration Examples

# Upgrade to latest version
ipk db upgrade

# Upgrade to specific revision
ipk db upgrade 001_initial

# Upgrade by one revision
ipk db upgrade +1

# Downgrade by one revision
ipk db downgrade -1

# Downgrade to base (WARNING: deletes all data)
ipk db downgrade base

# Show SQL without executing
ipk db upgrade --sql

3. Create Your FastAPI Application

from fastapi import FastAPI, Depends
from identity_plan_kit import (
    IdentityPlanKit,
    IdentityPlanKitConfig,
    CurrentUser,
)
from identity_plan_kit.auth.domain.entities import User

# Configure the library
config = IdentityPlanKitConfig()

# Create IdentityPlanKit instance
kit = IdentityPlanKit(
    config,
    startup_timeout=30.0,
    shutdown_drain_timeout=30.0,
)

# Create FastAPI app with lifespan
app = FastAPI(lifespan=kit.lifespan)

# Setup routes and middleware
kit.setup(
    app,
    register_error_handlers=True,
    include_health_routes=True,
    include_request_id=True,
)


# Use authentication in your routes
@app.get("/protected")
async def protected_route(user: User = Depends(CurrentUser(kit))):
    return {"message": f"Hello {user.email}"}


# Check user permissions
@app.get("/admin-only")
async def admin_only(user: User = Depends(CurrentUser(kit))):
    # Check if user has admin role
    has_permission = await kit.rbac_service.check_user_permission(
        user.id,
        "admin.access"
    )

    if not has_permission:
        raise HTTPException(status_code=403, detail="Forbidden")

    return {"message": "Admin access granted"}


# Track feature usage
@app.post("/generate")
async def generate(user: User = Depends(CurrentUser(kit))):
    # Check and track feature usage
    usage = await kit.plan_service.check_and_track_usage(
        user.id,
        "api_calls",
        increment=1,
    )

    if not usage.has_access:
        raise HTTPException(
            status_code=429,
            detail=f"Quota exceeded. Used {usage.usage}/{usage.limit}"
        )

    # Your feature logic here
    return {"remaining": usage.limit - usage.usage}

4. Run Your Application

uvicorn main:app --reload

Your app will have these endpoints automatically:

• GET /auth/google - Start Google OAuth flow
• GET /auth/google/callback - OAuth callback
• POST /auth/refresh - Refresh access token
• POST /auth/logout - Logout
• GET /health - Full health check
• GET /health/live - Liveness probe
• GET /health/ready - Readiness probe

Database Migrations

For Library Users

When you install identity-plan-kit, the library includes pre-built migrations in the alembic/ folder. You apply them using the CLI:

# Set your database URL
export IPK_DATABASE_URL=postgresql+asyncpg://user:pass@localhost/db

# Run migrations
ipk db upgrade

The initial migration (001_initial) creates all necessary tables:

• Authentication tables (users, providers, refresh_tokens)
• RBAC tables (roles, permissions, role_permissions)
• Plans tables (plans, features, limits, user_plans, usage)

Included Migrations

The library ships with:

1. 001_initial: Creates all core tables with:
   • Default roles: admin, user
   • Default plans: free, pro
   • All necessary indexes and constraints

Migration Workflow

# 1. Install the package
pip install identity-plan-kit

# 2. Set environment variables
export IPK_DATABASE_URL=postgresql+asyncpg://user:pass@localhost/mydb

# 3. Check current status
ipk db current

# 4. Apply migrations
ipk db upgrade

# 5. Verify
ipk db history

Troubleshooting

"IPK_DATABASE_URL environment variable not set"

export IPK_DATABASE_URL=postgresql+asyncpg://user:pass@localhost/db

"alembic.ini not found"

Make sure the package is properly installed:

pip install --force-reinstall identity-plan-kit

"Database connection failed"

Check your database is running and credentials are correct:

# Test connection
psql postgresql://user:pass@localhost/db -c "SELECT 1"

Migration already applied

If you've manually created tables, you can mark the database as migrated:

ipk db stamp head

Advanced Usage

Custom Cleanup Schedule

# Create a cleanup scheduler for expired tokens
scheduler = kit.create_cleanup_scheduler()

@asynccontextmanager
async def lifespan(app: FastAPI):
    await kit.startup()
    scheduler.start()  # Start background cleanup
    yield
    scheduler.stop()   # Stop cleanup
    await kit.shutdown()

app = FastAPI(lifespan=lifespan)

Health Checks

# Add custom health checks
kit.health_checker.register_check(
    "external_api",
    my_api_health_check,
    critical=False,
)

RBAC

# Check user permission
has_access = await kit.rbac_service.check_user_permission(
    user_id=user.id,
    permission_code="feature.access",
)

# Get user roles
roles = await kit.rbac_service.get_user_roles(user.id)

Plans and Usage

# Get user's active plan
plan = await kit.plan_service.get_user_active_plan(user.id)

# Check feature limit
usage_info = await kit.plan_service.check_and_track_usage(
    user.id,
    feature_code="api_calls",
    increment=1,
)

if usage_info.has_access:
    # User has quota remaining
    print(f"Remaining: {usage_info.limit - usage_info.usage}")
else:
    # Quota exceeded
    raise HTTPException(status_code=429, detail="Quota exceeded")

Configuration

All configuration is via environment variables or IdentityPlanKitConfig:

Variable	Description	Default
IPK_DATABASE_URL	PostgreSQL connection URL (async)	Required
IPK_SECRET_KEY	Secret key for JWT signing	Required
IPK_GOOGLE_CLIENT_ID	Google OAuth client ID	Required for OAuth
IPK_GOOGLE_CLIENT_SECRET	Google OAuth secret	Required for OAuth
IPK_GOOGLE_REDIRECT_URI	OAuth callback URL	Required for OAuth
IPK_ENVIRONMENT	Environment (development/production)	production
IPK_REDIS_URL	Redis URL for caching/sessions	Optional
IPK_ACCESS_TOKEN_EXPIRE_MINUTES	Access token TTL	15
IPK_REFRESH_TOKEN_EXPIRE_DAYS	Refresh token TTL	30
IPK_ENABLE_METRICS	Enable Prometheus metrics endpoint	false
IPK_METRICS_PATH	Path for metrics endpoint	/metrics
IPK_ENABLE_AUTO_CLEANUP	Auto cleanup expired tokens	true
IPK_CLEANUP_INTERVAL_HOURS	Cleanup interval in hours	6.0

Prometheus Metrics (Optional)

IdentityPlanKit supports Prometheus metrics for production observability.

Installation

# Install with metrics support
pip install identity-plan-kit[metrics]

Configuration

# Enable metrics via environment variable
IPK_ENABLE_METRICS=true
IPK_METRICS_PATH=/metrics  # optional, defaults to /metrics

Or in code:

config = IdentityPlanKitConfig(
    enable_metrics=True,
    metrics_path="/metrics",
)

Available Metrics

Metric	Type	Description
ipk_http_requests_total	Counter	Total HTTP requests by method, endpoint, status
ipk_http_request_duration_seconds	Histogram	Request latency (p50, p95, p99)
ipk_http_requests_in_progress	Gauge	Currently processing requests
ipk_auth_attempts_total	Counter	Auth attempts by provider and result
ipk_token_operations_total	Counter	Token operations (refresh, revoke, cleanup)
ipk_tokens_cleaned_total	Counter	Expired tokens cleaned up
ipk_circuit_breaker_state	Gauge	Circuit breaker state (0=closed, 1=open, 2=half_open)
ipk_db_connections_active	Gauge	Active database connections
ipk_quota_checks_total	Counter	Quota checks by feature and result
ipk_rate_limit_hits_total	Counter	Rate limit hits by endpoint
ipk_component_health_status	Gauge	Component health (1=healthy, 0=unhealthy)

Example Prometheus Scrape Config

scrape_configs:
  - job_name: 'identity-plan-kit'
    static_configs:
      - targets: ['localhost:8000']
    metrics_path: /metrics

Token Cleanup

Expired tokens are automatically cleaned up in the background (enabled by default).

# Disable auto cleanup
config = IdentityPlanKitConfig(
    enable_auto_cleanup=False,
)

# Or configure interval
config = IdentityPlanKitConfig(
    cleanup_interval_hours=12.0,  # Run every 12 hours
)

Manual cleanup is also available:

# Manual cleanup
deleted = await kit.cleanup_expired_tokens()
print(f"Cleaned up {deleted} expired tokens")

Development

# Clone repository
git clone https://github.com/yourusername/identity-plan-kit.git
cd identity-plan-kit

# Install with dev dependencies
uv pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy src

# Linting
ruff check src

License

MIT License - see LICENSE file for details.

Contributing

Contributions welcome! Please open an issue or submit a PR.