nilauth-credit-middleware 0.1.2

nilauth-credit-middleware

FastAPI Credit Locking Middleware

A comprehensive middleware solution for FastAPI that integrates with the nilauth-credit service to implement pay-per-request functionality.

Features

• 🔒 Automatic credit locking before request processing
• 💰 Dynamic cost calculation based on actual resource usage
• 🎯 Selective metering - only affect specific endpoints
• 🔄 Automatic rollback on request failures
• 🛠️ Flexible configuration - multiple user ID extraction methods
• 📊 Built-in cost calculators for common scenarios

Installation

pip install -r requirements.txt

Quick Start

1. Basic Setup

from fastapi import FastAPI, Request
from nilauth_credit_middleware import (
    CreditClientSingleton,
    metered,
    MeteringConfig
)

app = FastAPI()

# Configure the singleton credit client at startup
CreditClientSingleton.configure(
    base_url="http://localhost:3000",
    api_token="your-api-token",
    timeout=10.0
)

2. Create a User ID Extractor

async def get_user_id(request: Request) -> str:
    """Extract user ID from request header"""
    user_id = request.headers.get("X-User-ID")
    if not user_id:
        raise ValueError("User ID header not found")
    return user_id

3. Decorate Your Endpoints

@app.get("/api/expensive-operation")
@metered(
    config=MeteringConfig(
        user_id_extractor=get_user_id,
        credential="cred_user1_private",  # User's credential key
        estimated_cost=5.0,
        use_public_endpoint=False  # Use private lock endpoint
    )
)
async def expensive_operation(request: Request):
    return {"result": "success"}

Note: The API now requires a credential key for locking funds. Credentials can be either:

• Private credentials: Use with use_public_endpoint=False for internal operations
• Public credentials: Use with use_public_endpoint=True for user-facing endpoints

Usage Examples

Example 1: Fixed Cost Endpoint

Charge a fixed amount regardless of processing:

@app.get("/api/simple")
@metered(
    config=MeteringConfig(
        user_id_extractor=get_user_id,
        credential="cred_user1_private",
        estimated_cost=1.0,  # Will charge exactly 1.0 credit
        use_public_endpoint=False
    )
)
async def simple_endpoint(request: Request):
    return {"message": "Fixed cost endpoint"}

Example 2: Dynamic Cost Based on Response Size

from nilauth_credit_middleware import CostCalculators, MeteringConfig

@app.post("/api/data")
@metered(
    config=MeteringConfig(
        user_id_extractor=get_user_id,
        credential="cred_user1_public",  # Using public credential
        estimated_cost=5.0,  # Lock 5.0 credits upfront
        cost_calculator=CostCalculators.by_response_size(
            base_cost=0.5,    # Base cost
            per_kb=0.1        # Additional cost per KB
        ),
        use_public_endpoint=True  # Use public lock endpoint
    )
)
async def get_data(request: Request):
    # Return large data
    return {"data": [...]}  # Cost = 0.5 + (size_in_kb * 0.1)

Example 3: Cost Based on Processing Time

@app.post("/api/process")
@metered(
    config=MeteringConfig(
        user_id_extractor=get_user_id,
        credential="cred_user1_private",
        estimated_cost=10.0,
        cost_calculator=CostCalculators.by_processing_time(
            base_cost=1.0,     # Base cost
            per_second=2.0     # Cost per second
        )
    )
)
async def process_data(request: Request):
    request.state.start_time = time.time()
    # Do processing...
    return {"result": "done"}  # Cost = 1.0 + (duration * 2.0)

Example 4: Custom Cost Calculation

async def custom_calculator(request: Request, response: Response) -> float:
    """Calculate cost based on custom business logic"""
    base_cost = 1.0
    
    # Factor 1: Response size
    if hasattr(response, 'body'):
        size_cost = len(response.body) / 1024.0 * 0.05
    else:
        size_cost = 0.0
    
    # Factor 2: Processing time
    start_time = getattr(request.state, 'start_time', None)
    if start_time:
        duration = time.time() - start_time
        time_cost = duration * 1.0
    else:
        time_cost = 0.0
    
    # Factor 3: Success/failure penalty
    error_penalty = 0.5 if response.status_code >= 400 else 0.0
    
    return base_cost + size_cost + time_cost + error_penalty

@app.post("/api/advanced")
@metered(
    config=MeteringConfig(
        user_id_extractor=get_user_id,
        credential="cred_user1_private",
        estimated_cost=15.0,
        cost_calculator=custom_calculator
    )
)
async def advanced_endpoint(request: Request):
    request.state.start_time = time.time()
    # Process request...
    return {"result": "processed"}

User ID Extraction Methods

From Header

async def from_header(request: Request) -> str:
    user_id = request.headers.get("X-User-ID")
    if not user_id:
        raise ValueError("X-User-ID header required")
    return user_id

From Query Parameter

async def from_query(request: Request) -> str:
    user_id = request.query_params.get("user_id")
    if not user_id:
        raise ValueError("user_id parameter required")
    return user_id

From JWT Token

async def from_jwt(request: Request) -> str:
    import jwt
    
    auth = request.headers.get("Authorization")
    if not auth or not auth.startswith("Bearer "):
        raise ValueError("Bearer token required")
    
    token = auth.split(" ")[1]
    payload = jwt.decode(token, options={"verify_signature": False})
    return payload.get("sub")

From Request Body

from pydantic import BaseModel

class RequestWithUser(BaseModel):
    user_id: str
    data: dict

async def from_body(request: Request) -> str:
    body = await request.json()
    user_id = body.get("user_id")
    if not user_id:
        raise ValueError("user_id not found in body")
    return user_id

Built-in Cost Calculators

Fixed Cost

CostCalculators.fixed_cost(amount=2.5)
# Always charges 2.5 credits

By Response Size

CostCalculators.by_response_size(
    base_cost=0.1,  # Base cost
    per_kb=0.01     # Cost per kilobyte
)
# Cost = base_cost + (response_size_kb * per_kb)

By Processing Time

CostCalculators.by_processing_time(
    base_cost=0.1,   # Base cost
    per_second=0.5   # Cost per second
)
# Cost = base_cost + (duration_seconds * per_second)
# Requires: request.state.start_time = time.time()

Error Handling

The middleware automatically handles errors:

1. Lock fails (insufficient balance, user not found, etc.)

   • Request is rejected immediately
   • No processing occurs
   • Client receives appropriate error

2. Processing fails (exception during request handling)

   • Funds are unlocked with 0 cost (full refund)
   • Original exception is propagated

3. Unlock fails (network error, etc.)

   • Error is logged
   • Original response is still returned
   • Manual intervention may be needed

Running the Example

1. Start the nilauth-credit service:

# From the main project directory
docker-compose up -d
cargo run

2. Start the Python example app:

cd python_middleware
python example_app.py

3. Test the endpoints:

# Simple endpoint (1.0 credit)
curl -H "X-User-ID: test-user-1" http://localhost:8000/api/simple

# Variable response size
curl -X POST -H "X-User-ID: test-user-1" \
  -H "Content-Type: application/json" \
  -d '{"size": 1000}' \
  http://localhost:8000/api/data

# Processing time-based
curl -X POST -H "X-User-ID: test-user-1" \
  "http://localhost:8000/api/process?duration=0.5"

# Check balance
curl "http://localhost:8000/api/test/balance?user_id=test-user-1"

API Documentation

Once running, visit:

• Interactive docs: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

Configuration Options

CreditClientSingleton

CreditClientSingleton.configure(
    base_url="http://localhost:3000",  # Credit service URL
    api_token="your-token",            # API authentication token (admin key)
    timeout=10.0                       # Request timeout in seconds
)

@metered Decorator

@metered(
    config=MeteringConfig(
        user_id_extractor=callable,     # Required: async function to get user_id
        credential="cred_xxx",           # Required: user credential key
        estimated_cost=1.0,              # Required: upfront lock amount
        cost_calculator=callable,        # Optional: async function for actual cost
        use_public_endpoint=False,       # Optional: use public lock endpoint (default: False)
        default_cost_percentage=0.1      # Optional: default cost if calculator fails (default: 0.1)
    )
)

Credentials

The API uses two types of credentials:

• Private credentials (cred_user1_private): For internal/private operations
  • Use with use_public_endpoint=False
  • Lock via /v1/balance/lock/private
• Public credentials (cred_user1_public): For user-facing operations
  • Use with use_public_endpoint=True
  • Lock via /v1/balance/lock/public

Testing

# Install test dependencies
pip install pytest pytest-asyncio

# Run tests (if available)
pytest

Best Practices

1. Estimate conservatively: Set estimated_cost high enough to cover most cases
2. Calculate accurately: Use cost calculators to charge fair amounts
3. Handle errors gracefully: Always validate user IDs and handle missing data
4. Monitor closely: Log cost calculations for debugging
5. Secure tokens: Never commit API tokens to version control
6. Use environment variables: Load configuration from env vars in production

Integration with Your Service

Environment Variables

export CREDIT_SERVICE_URL="http://localhost:3000"
export CREDIT_API_TOKEN="your-api-token-here"

Production Configuration

import os
from nilauth_credit_middleware import CreditClientSingleton

# Configure at app startup
CreditClientSingleton.configure(
    base_url=os.getenv("CREDIT_SERVICE_URL", "http://localhost:3000"),
    api_token=os.getenv("CREDIT_API_TOKEN"),  # Admin API token
    timeout=float(os.getenv("CREDIT_TIMEOUT", "10.0"))
)

# Use environment variables for credentials too
USER_PRIVATE_CREDENTIAL = os.getenv("USER_PRIVATE_CREDENTIAL", "cred_user1_private")
USER_PUBLIC_CREDENTIAL = os.getenv("USER_PUBLIC_CREDENTIAL", "cred_user1_public")

Troubleshooting

"Credit service unavailable"

• Ensure nilauth-credit service is running
• Check base_url is correct
• Verify network connectivity

"User not found"

• Verify user exists in database
• Check user ID extraction logic
• Review API token permissions

"Insufficient balance"

• User needs to top up credits
• Check current balance via /v1/summary endpoint
• Consider reducing estimated costs

"Lock not found"

• May indicate timing issue
• Check for race conditions
• Review unlock error logs

License

Same as the nilauth-credit project.

Support

For issues and questions, please refer to the main nilauth-credit project documentation.