pancake-framework 0.1.3

Pancake - Decorator-driven Python web framework with IoC, MyBatis ORM, and AI workflow

Pancake Framework

A decorator-driven Python web framework with IoC, MyBatis-style ORM, and AI workflow integration.

中文文档

Features

• Zero Import - All decorators and services auto-injected into builtins, no import needed
• Decorator-Driven - Register services, controllers, and mappers with simple decorators
• CLI Tool - pancake create/run/check/build commands for project management
• Auto Dependency Injection - @auto_inject() automatically resolves parameters from YAML/JSON config
• IoC Container - Singleton, transient, and scoped dependency management
• MyBatis Plus ORM - Async ORM with BaseMapper CRUD, @Select/@Insert SQL annotations, dynamic SQL, chain queries
• Multi-Database - SQLite / PostgreSQL / MySQL with auto-detection
• FastAPI Web Server - Built-in @get_controller/@post_controller and all HTTP methods
• Auth & Authorization - @auth_required, @role_required, pluggable auth handlers
• Middleware & Validation - @middleware, @validate, @transaction decorators
• AI Module - Unified LLM client (OpenAI/DeepSeek/Gemini/Ollama), short-term & long-term memory, RAG
• LangGraph Integration - AI workflow nodes, edges, and state graphs
• Redis Cache - @cached decorator with anti-penetration/avalanche/breakdown protection
• Message Queue - In-memory SimpleBroker and RedisBroker for event-driven architecture
• Remote Calls - @remote_node for HTTP and gRPC remote invocation
• Lifecycle Management - Lifecycle base class with init/start/stop/error hooks
• CUI - Click-based CLI command registration with @cui_command
• GUI - Flet (Flutter) based GUI page registration with @gui_page
• Plugin System - Auto-discovery with init-order control, external plugin dirs
• Centralized Settings - All paths and configs managed through settings.py
• Database Migration - Version-based schema migration with auto-detect changes
• Database Dialects - Automatic SQLite / PostgreSQL / MySQL type mapping and syntax
• Connection Health - Ping, auto-reconnect for database connections
• Rate Limiting - @rate_limit(times, seconds) per-IP sliding window
• WebSocket - @websocket_controller for real-time communication
• Config Hot-reload - File watcher for YAML/JSON config changes
• Metrics - /metrics endpoint with request count, latency, error rate
• API Docs - Auto-generated Swagger (/docs) and ReDoc (/redoc)

Quick Start

Install

pip install pancake_framework

Create a Project

pancake create myapp
cd myapp

Run

# Using CLI
pancake run

# Or using Python
python main.py

The server starts at http://127.0.0.1:8080 by default. Health check at /health.

CLI Commands

Command	Description
pancake create <name>	Create a new project with standard structure
pancake run	Run the project
pancake check	Check project structure and environment
pancake build	Package project as wheel

Usage

Web Controller (no import needed)

@get_controller("/hello")
def hello():
    return {"message": "Hello from Pancake!"}

@post_controller("/users")
async def create_user(name: str, age: int):
    return {"id": await UserMapper().insert(name=name, age=age)}

Auth & Authorization

@set_auth_handler
async def authenticate(request, token):
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401)
    return user

@get_controller("/profile")
@auth_required
async def get_profile(current_user):
    return {"user": current_user}

@delete_controller("/admin/users/{user_id}")
@role_required("admin")
async def delete_user(user_id: int):
    await UserMapper().delete_by_id(user_id)

Middleware & Transaction

@middleware(order=1)
async def log_request(request, call_next):
    start = time.time()
    response = await call_next(request)
    logger.info(f"{request.method} {request.url.path} {time.time()-start:.3f}s")
    return response

@post_controller("/transfer")
@transaction
async def transfer(from_id: int, to_id: int, amount: float):
    # All DB operations in this function run in a single transaction
    ...

MyBatis Plus ORM (no import needed)

@Mapper
class UserMapper(BaseMapper):
    @dataclass
    class User:
        id: int = None
        name: str = None
        age: int = None

    _entity_class = User
    _table_name = "users"

    @Select("SELECT * FROM users WHERE name = #{name}")
    async def find_by_name(self, name: str) -> list[User]: ...

Built-in CRUD: select_by_id, select_list, select_one, select_count, insert, insert_batch, update_by_id, delete_by_id.

Chain queries:

users = await mapper.select(qw().ge("age", 18).like("name", "%Ali%").order_by_desc("age").limit(50))
await mapper.update(uw().set("name", "Bob").eq("id", 1))
await mapper.delete(qw().lt("age", 18))

AI Module (no import needed)

Configure src/resource/yaml/ai.yaml, then use directly:

# Chat
response = await chat_model.chat([{"role": "user", "content": "Hello"}])

# Stream
async for chunk in chat_model.chat_stream([...]):
    print(chunk, end="")

# Short-term memory (session context)
await short_term_memory.add("session_001", "user", "My name is Alice")
messages = await short_term_memory.get_messages("session_001")

# Long-term memory (persistent)
await long_term_memory.remember("user_name", "Alice")
name = await long_term_memory.recall("user_name")

# RAG
await rag.add_document("Pancake is a Python framework...")
answer = await rag.ask("What is Pancake?")

Supported providers: OpenAI, DeepSeek, Gemini, Ollama, GLM, Moonshot, Qwen, vLLM.

Redis Cache

@cached(ttl=300)
async def get_user(user_id: int):
    return await db.query(user_id)

# CacheGuard with anti-penetration/avalanche/breakdown
guard = CacheGuard(redis_client)
user = await guard.get_or_load("user:123", lambda: db.query(123), ttl=600, jitter=60)

Event-Driven Messaging

@event_node(name="order_created", event="order.created")
async def create_order(item: str, qty: int):
    return {"item": item, "qty": qty, "status": "created"}

@on_event("order.created")
async def notify_inventory(message):
    print(f"Order received: {message}")

Lifecycle Hooks

class MyService(Lifecycle):
    async def on_init(self):
        self.cache = {}

    async def on_start(self):
        await self.load_data()

    async def on_stop(self):
        await self.cleanup()

CUI (CLI Commands)

@cui_command("greet", help="Say hello")
@cui_option("--name", "-n", default="World", help="Name")
def greet(name: str):
    click.echo(f"Hello, {name}!")

GUI (Flet/Flutter)

@gui_page("/", title="Home")
def home(page: ft.Page):
    page.add(ft.Text("Welcome to Pancake GUI"))

Optional Dependencies

pip install pancake_framework[ai]          # AI module (OpenAI, Gemini, etc.)
pip install pancake_framework[langgraph]   # LangGraph AI workflow
pip install pancake_framework[redis]       # Redis cache and message queue
pip install pancake_framework[grpc]        # gRPC remote calls
pip install pancake_framework[cui]         # Click CLI commands
pip install pancake_framework[gui]         # Flet GUI
pip install pancake_framework[all]         # All optional deps

TODO

• Database migration support
• Configuration hot-reload
• Pagination Page object abstraction
• OpenTelemetry / metrics integration
• Graceful shutdown with signal handling
• WebSocket support
• Rate limiting middleware
• API documentation auto-generation
• More database dialects (SQLite/PG/MySQL type mapping)
• Connection pool health check and auto-reconnect

Performance

• C extension — convert hot-path modules (sql_parser, wrapper, jwt) to C for speed

Running Tests

pip install pytest pytest-asyncio
python -m pytest tests/ -v

License

MIT