Skip to content

owasp-sbot/OSBot-Fast-API

Repository files navigation

OSBot-Fast-API

Current Release Python FastAPI Type-Safe AWS Lambda

A Type-Safe wrapper around FastAPI that provides strong typing, comprehensive middleware support, HTTP event tracking, and seamless AWS Lambda integration through Mangum.

✨ Key Features

  • πŸ” Type-Safe First: Automatic bidirectional conversion between Type_Safe classes and Pydantic BaseModels
  • πŸ›‘οΈ Built-in Middleware: API key validation, CORS, disconnect detection, and HTTP event tracking
  • πŸ“Š HTTP Event System: Comprehensive request/response tracking with configurable storage
  • πŸš€ AWS Lambda Ready: Direct integration with Mangum for serverless deployment
  • πŸ§ͺ Testing Utilities: Built-in test server with Type-Safe support
  • πŸ”„ Auto-conversion: Seamless Type_Safe ↔ BaseModel ↔ Dataclass conversions
  • πŸ“ Route Organization: Clean route structure with automatic path generation

πŸ“¦ Installation

pip install osbot-fast-api

πŸš€ Quick Start

Basic Application

from osbot_fast_api.api.Fast_API import Fast_API
from osbot_fast_api.api.routes.Fast_API__Routes import Fast_API__Routes
from osbot_utils.type_safe.Type_Safe import Type_Safe


# Define Type-Safe schema
class User(Type_Safe):
    username: str
    email: str
    age: int


# Create routes
class Routes_Users(Fast_API__Routes):
    tag = 'users'

    def create_user(self, user: User):
        # user is automatically converted from BaseModel to Type_Safe
        return {'created': user.username}

    def get_user__id(self, id: str):  # Becomes /users/get-user/{id}
        return {'user_id': id}

    def setup_routes(self):
        self.add_route_post(self.create_user)
        self.add_route_get(self.get_user__id)


# Setup application
fast_api = Fast_API(enable_cors=True)
fast_api.setup()
fast_api.add_routes(Routes_Users)

# Get FastAPI app instance
app = fast_api.app()

With Middleware & Authentication

import os

# Configure API key authentication
os.environ['FAST_API__AUTH__API_KEY__NAME'] = 'X-API-Key'
os.environ['FAST_API__AUTH__API_KEY__VALUE'] = 'your-secret-key'

# Create app with middleware
fast_api = Fast_API(
    enable_cors=True,      # Enable CORS support
    enable_api_key=True,   # Enable API key validation
    default_routes=True    # Add /status, /version routes
)

# Configure HTTP event tracking
fast_api.http_events.max_requests_logged = 100
fast_api.http_events.clean_data = True  # Sanitize sensitive headers

fast_api.setup()

πŸ—οΈ Architecture

OSBot-Fast-API extends FastAPI with a comprehensive Type-Safe layer and monitoring capabilities:

β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚                   Your Application                  β”‚
β”‚                                                     β”‚
β”‚  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”   β”‚
β”‚  β”‚ Type-Safe    β”‚  β”‚   Routes     β”‚  β”‚  Events  β”‚   β”‚
β”‚  β”‚  Schemas     β”‚  β”‚   Classes    β”‚  β”‚ Handlers β”‚   β”‚
β”‚  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜   β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                        β”‚
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚                  OSBot-Fast-API                     β”‚
β”‚                                                     β”‚
β”‚  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”     β”‚
β”‚  β”‚          Type Conversion System            β”‚     β”‚
β”‚  β”‚   Type_Safe ↔ BaseModel ↔ Dataclass        β”‚     β”‚
β”‚  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜     β”‚
β”‚                                                     β”‚
β”‚  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”     β”‚
β”‚  β”‚           Middleware Pipeline              β”‚     β”‚
β”‚  β”‚  Disconnect β†’ Events β†’ CORS β†’ API Key      β”‚     β”‚
β”‚  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜     β”‚
β”‚                                                     β”‚
β”‚  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”     β”‚
β”‚  β”‚         HTTP Event Tracking System         β”‚     β”‚
β”‚  β”‚   Request/Response/Traces/Monitoring       β”‚     β”‚
β”‚  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜     β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                        β”‚
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚                     FastAPI                         β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜

πŸ” Type-Safe Integration

OSBot-Fast-API automatically converts between Type_Safe classes and Pydantic BaseModels:

from osbot_utils.type_safe.Type_Safe import Type_Safe
from typing import List, Optional

# Define Type-Safe schemas (not Pydantic!)
class Address(Type_Safe):
    street: str
    city: str
    country: str

class Person(Type_Safe):
    name: str
    age: int
    email: Optional[str] = None
    addresses: List[Address] = []

# Use directly in routes - automatic conversion happens
class Routes_People(Fast_API__Routes):
    tag = 'people'
    
    def create_person(self, person: Person):
        # person is Type_Safe instance, not BaseModel
        # Full type validation and conversion handled automatically
        return person  # Automatically converted back to JSON
    
    def setup_routes(self):
        self.add_route_post(self.create_person)

πŸ“Š HTTP Event Tracking

Built-in comprehensive request/response tracking:

# Configure event tracking
fast_api.http_events.max_requests_logged = 100
fast_api.http_events.clean_data = True  # Sanitize sensitive data
fast_api.http_events.trace_calls = True  # Enable execution tracing (debug)

# Add event callbacks
def on_request(event):
    print(f"Request: {event.http_event_request.path}")

def on_response(response, event):
    print(f"Response: {event.http_event_response.status_code}")
    print(f"Duration: {event.http_event_request.duration}s")

fast_api.http_events.callback_on_request = on_request
fast_api.http_events.callback_on_response = on_response

πŸ›‘οΈ Middleware Stack

Built-in middleware pipeline (in execution order):

  1. Detect_Disconnect: Monitor client disconnections
  2. Http_Request: Event tracking and logging
  3. CORS: Cross-origin resource sharing
  4. API_Key_Check: Header/cookie API key validation

Custom Middleware

class Custom_Fast_API(Fast_API):
    def setup_middlewares(self):
        super().setup_middlewares()  # Add default middleware
        
        @self.app().middleware("http")
        async def add_process_time(request: Request, call_next):
            import time
            start = time.time()
            response = await call_next(request)
            response.headers["X-Process-Time"] = str(time.time() - start)
            return response

πŸ§ͺ Testing

Built-in test server with Type-Safe support:

from osbot_fast_api.utils.Fast_API_Server import Fast_API_Server

def test_api():
    fast_api = Fast_API()
    fast_api.setup()
    fast_api.add_routes(Routes_Users)
    
    with Fast_API_Server(app=fast_api.app()) as server:
        # Test with Type-Safe object
        user_data = {'username': 'alice', 'email': '[email protected]', 'age': 30}
        response = server.requests_post('/users/create-user', data=user_data)
        
        assert response.status_code == 200
        assert response.json()['created'] == 'alice'

πŸš€ AWS Lambda Deployment

from mangum import Mangum
from osbot_fast_api.api.Fast_API import Fast_API

# Create and setup application
fast_api = Fast_API()
fast_api.setup()
fast_api.add_routes(Routes_Users)

# Create Lambda handler
app = fast_api.app()
handler = Mangum(app)

def lambda_handler(event, context):
    return handler(event, context)

πŸ“ Project Structure

osbot_fast_api/
β”œβ”€β”€ api/
β”‚   β”œβ”€β”€ Fast_API.py                 # Main FastAPI wrapper
β”‚   β”œβ”€β”€ Fast_API__Routes.py          # Route organization base class
β”‚   β”œβ”€β”€ Fast_API__Http_Event*.py    # Event tracking components
β”‚   └── middlewares/                # Built-in middleware
β”œβ”€β”€ utils/
β”‚   β”œβ”€β”€ type_safe/                  # Type conversion system
β”‚   β”‚   β”œβ”€β”€ Type_Safe__To__BaseModel.py
β”‚   β”‚   β”œβ”€β”€ BaseModel__To__Type_Safe.py
β”‚   β”‚   └── ...
β”‚   β”œβ”€β”€ Fast_API_Server.py          # Test server
β”‚   └── Fast_API_Utils.py           # Utilities
└── examples/                        # Usage examples

πŸ“š Documentation

Comprehensive documentation is available in the /docs folder:

🎯 Key Benefits

For Developers

  • Type Safety: Catch errors at development time with Type_Safe validation
  • Less Boilerplate: Convention over configuration approach
  • Auto-conversion: Seamless type conversions at API boundaries
  • Built-in Testing: Integrated test server and utilities

For Production

  • Monitoring: Comprehensive HTTP event tracking
  • Security: Built-in API key validation and header sanitization
  • Performance: Cached type conversions and efficient middleware
  • AWS Ready: Direct Lambda integration with Mangum

For Teams

  • Organized Code: Clear separation with Fast_API__Routes classes
  • Consistent Patterns: Standardized route naming and structure
  • Easy Testing: Type-Safe test utilities
  • Documentation: Auto-generated OpenAPI/Swagger docs

πŸ”§ Advanced Features

Route Path Generation

  • get_users() β†’ /get-users
  • get_user__id() β†’ /get-user/{id}
  • user__id_posts__post_id() β†’ /user/{id}/posts/{post_id}

Type-Safe Primitives

from osbot_utils.type_safe.Type_Safe__Primitive import Type_Safe__Primitive

class Email(Type_Safe__Primitive, str):
    def __new__(cls, value):
        if '@' not in value:
            raise ValueError("Invalid email")
        return super().__new__(cls, value)

Event Access in Routes

from fastapi import Request

def get_request_info(self, request: Request):
    return {
        'event_id': str(request.state.request_id),
        'thread_id': request.state.request_data.http_event_info.thread_id
    }

🀝 Contributing

Contributions are welcome! Please check the documentation for architecture details and patterns.

πŸ“„ License

This project is licensed under the Apache 2.0 License.

πŸ”— Related Projects

πŸ’‘ Examples

For more examples, see:


Built with ❀️ using Type-Safe principles for robust, maintainable APIs

About

OSBot helpers for FastAPI

Resources

License

Stars

Watchers

Forks

Packages

No packages published

Contributors 2

  •  
  •