py-aide 0.16.0

Modern Python 3.11+ Framework: Flask/FastAPI orchestration, and strict runtime enforcement.

py_aide

Modern Python 3.11+ Framework: Flask/FastAPI Orchestration & Strict Runtime Enforcement

py_aide is a robust, developer-centric framework designed to bring safety, structure, and consistency to Python web applications. It provides a unique "Strict Runtime Enforcement" layer that ensures your code remains clean, documented, and type-safe at execution time.

🚀 Key Features

• Strict Runtime Enforcement: A powerful decorator that enforces type hints, docstrings, and calling conventions (positional-only/keyword-only) at runtime.
• Unified Server Portal: Seamlessly orchestrate Flask and FastAPI applications with shared security gates and auto-discovery of routes.
• WebSocket Excellence: High-performance, identity-aware WebSockets with support for User and Group-based messaging, automatic handshake mapping, and tiered delivery.
• Thread-Safe SQL: A thread-level multiton SQLite manager with automatic JSON serialization, transaction tracking, and schema management.
• Modern Security: First-class support for Bearer tokens, API keys, and Fernet-based encryption.
• Enterprise HTTP Client: A standardized, "always resolve" HTTP client with built-in retries, interceptors, and dual sync/async support.
• Rich Utilities: Built-in handlers for images (base64/files), dates (aware/naive conversions), and custom data structures.

📦 Installation

pip install py_aide

Note: py_aide requires Python 3.11+ and is currently optimized for Linux environments.

⚙️ Core Philosophy: Strict Enforcement

At the heart of py_aide is the @enforce_requirements decorator. It's designed to prevent "sloppy" code by failing early if:

• A function is missing a docstring.
• A parameter or return value is missing a type hint.
• A function uses more than 8 arguments (promoting better decomposition).
• Calling conventions (/ or *) are not explicitly defined.

from py_aide.enforcer import enforce_requirements

@enforce_requirements
def create_user(name: str, age: int, /) -> dict:
    """Creates a new user dict."""
    return {"name": name, "age": age}

🌐 Unified Server Example (Flask)

from py_aide.servers.flask import ServicePortal, GateConfig

portal = ServicePortal()

@portal.endpoint("/api/greet", gate=GateConfig(auth_required=False))
def greet(name: str, /) -> dict:
    """Returns a greeting message."""
    return {"message": f"Hello, {name}!"}

if __name__ == "__main__":
    portal.run(port=5000)

📡 Standardized HTTP Client (requests)

py_aide provides a resilient HTTP client designed to work across both Flask and FastAPI. It follows an "Always Resolve" philosophy: it never raises exceptions for network errors or HTTP failures. Instead, it returns a standardized Response object.

Core Features:

• Resilient: Automatic retries with exponential, linear, or fixed backoff.
• Secure: Auto-injects Bearer tokens and sanitizes outgoing payloads.
• Silent Mode: Suppress internal framework logging for specific requests.
• Middleware: Register global request and response interceptors.
• Dual-Mode: Dedicated paths for Synchronous (Flask) and Asynchronous (FastAPI) logic.

1. Synchronous Usage (Flask)

Ideal for standard routes where you want simple, linear logic.

from py_aide import send_request, RequestConfig

# Optional: Set global defaults
RequestConfig.base_url = "https://api.myapp.com"
RequestConfig.max_retries = 3

def get_data():
    # Parameters are now keyword-only for strictness
    res = send_request(
        method="GET", 
        endpoint="/external-data",
        headers={"X-Custom-Header": "Value"}, # Custom headers
        silent=True, # Suppress all framework logs for this request
        on_success=lambda data: print("Got data!"),
        on_error=lambda err: print(f"Error: {err}")
    )
    
    if res:
        print(f"Success: {res.data}")
    else:
        print(f"Failed: {res.log}") # Detailed error log

2. Asynchronous Usage (FastAPI)

Optimized for high-concurrency environments using asyncio.

from py_aide import send_request_async

async def fetch_profile(user_token: str):
    # Automatically injects Bearer token
    res = await send_request_async(
        method="GET", 
        endpoint="/profile", 
        auth_token=user_token
    )
    return res.to_dict()

3. Interceptors (Middleware)

Globally modify requests before they leave or responses before they reach your logic.

from py_aide import Interceptors

# Add a header to every outgoing request
Interceptors.add_request_hook(lambda data: {
    **data, 
    "headers": {**data["headers"], "X-Client-ID": "py-aide-v1"}
})

# Log or transform every response
def log_response(res):
    print(f"Response from {res.errorLogs.get('endpoint')}")
    return res

Interceptors.add_response_hook(log_response)

4. Authentication Strategies

py_aide handles common auth patterns out-of-the-box.

# 1. Bearer Token (Mandatory auth_type)
send_request(method="GET", endpoint="/", auth_token="my-token", auth_type="bearer")

# 2. API Key (Uses X-API-Key header by default)
send_request(method="GET", endpoint="/", auth_token="sk_123", auth_type="api_key")

# 3. Basic Auth (Automatically Base64 encodes)
send_request(method="GET", endpoint="/", auth_token=("user", "pass"), auth_type="basic")

# Custom API Key Header
RequestConfig.api_key_header = "X-My-Custom-Auth"

📦 Persistent Queuing

py_aide includes a robust, SQLite-backed persistent queue for tasks that must survive application restarts.

from py_aide import PersistentQueue

# 1. Initialize (Defaults to storage/queues.db)
queue = PersistentQueue(queue_name="email_tasks")

# 2. Push a task
queue.push({"to": "[e-mail]", "body": "Hello!"}, priority=10)

# 3. Pop and Process
res = queue.pop()
if res:
    task = res.data[0]
    task_id = task["id"]
    payload = task["payload"]
    
    try:
        # Process task...
        queue.complete(task_id)
    except Exception as e:
        queue.fail(task_id, err=str(e), retry=True)

# 4. Peek without locking
upcoming = queue.peek(limit=5)

⚡ Real-Time Identity & Groups (WebSockets)

py_aide moves beyond anonymous broadcasts. It allows you to target users and groups directly using their business IDs (e.g., userId), handling the underlying session mapping automatically. This powerful API is unified across both FastAPI and Flask.

Unified Delivery Methods

Both FastAPIServicePortal and ServicePortal (Flask) expose identical methods for targeted communication:

• send_to_user(user_id, event, data): Reach all active sessions of a specific user.
• send_to_group(group_id, event, data): Sync messages within a room or collaborative group.
• broadcast(event, data): Send a message to every connected client.

Flask Example

from py_aide.servers.flask import ServicePortal
from py_aide.servers.gate import AuthType

portal = ServicePortal(enable_websocket=True)

@portal.on_event("join_team", auth_required=True, auth_type=AuthType.BEARER)
def handle_join(data: dict):
    """Adds the user to a collaborative group."""
    team_id = data.get("teamId")
    # Identity mapping is handled automatically upon successful event auth
    portal.join_group(team_id) 
    return {"status": "success", "team": team_id}

# Sending targeted messages from anywhere (even sync contexts)
def notify_user(user_id: str, message: str):
    portal.send_to_user(user_id, "notification", {"text": message})

FastAPI Example

from py_aide.servers.fastApi import FastAPIServicePortal

portal = FastAPIServicePortal(enable_websocket=True)

@portal.on_event("join_team", auth_required=True)
async def handle_join(data: dict, sid: str):
    team_id = data.get("teamId")
    await portal.join_group(sid, team_id)
    return {"status": "success"}

async def sync_team(team_id: str, update: dict):
    await portal.send_to_group(team_id, "team_update", update)

🗄️ Database Management

py_aide provides a thread-local multiton pattern for SQLite, ensuring each thread has its own connection while sharing the same configuration.

from py_aide.database import Api

db_config = {
    'users': 'id INTEGER PRIMARY KEY, name TEXT, meta JSON'
}

with Api(db_path="data.db", tables=db_config) as db:
    db.insert(table="users", data=[(1, "Alice", {"role": "admin"})])
    result = db.fetch(table="users", columns=["name", "meta::role as role"])
    print(result.data) # [{'name': 'Alice', 'role': 'admin'}]

⚠️ Important Note: Eventlet Monkey Patching

By default, importing src.py_aide (or the top-level package) immediately performs eventlet.monkey_patch(). This is required for reliable WebSocket support and some threading features. If you need to avoid this side-effect, ensure you understand the dependencies of your modules.

📄 License

This project is licensed under the MIT License. See the LICENSE file for more details.

👥 Authors

• Kakuru Douglas - [e-mail]