py-aide 0.14.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.
• 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)

⚡ 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]