Revolutionary Hybrid Async/Sync Web Framework for Python
β‘ The first framework with seamless sync/async support and reactive architecture
QakeAPI is a completely new approach to web frameworks:
- π Hybrid Sync/Async β write sync and async code simultaneously
- β‘ Reactive Routing β reactive routing and events
- π Parallel Dependencies β automatic dependency parallelism
- π Pipeline Composition β function composition into pipelines
- π― Smart Routing β intelligent routing based on conditions
- β Zero Dependencies β only Python standard library
- β Production-Ready β ready for real-world projects
- β Performance β automatic parallelism, optimized routing (Trie-based)
- β Simplicity β intuitive syntax
- β Flexibility β simultaneous sync and async support
- β OpenAPI/Swagger β automatic API documentation
- β WebSocket Support β real-time communication
- β Background Tasks β asynchronous task processing
- β Middleware System β customizable request/response processing
- β CORS Support β built-in CORS middleware
- β Dependency Injection β clean architecture with DI
- β Rate Limiting β built-in rate limiting decorator
- β Caching β response caching with TTL
- β Request Validation β automatic data validation
- β File Upload β multipart file upload with validation
- β Security β request size limits, validation, error handling
pip install qakeapifrom qakeapi import QakeAPI, CORSMiddleware
app = QakeAPI(
title="My API",
version="1.2.0",
description="My awesome API"
)
# Add CORS middleware
app.add_middleware(CORSMiddleware(allow_origins=["*"]))
# Sync function works automatically!
@app.get("/users/{id}")
def get_user(id: int):
return {"id": id, "name": f"User {id}"}
# Async function is also supported
@app.get("/items/{item_id}")
async def get_item(item_id: int):
return {"item_id": item_id}
# POST with automatic body extraction
@app.post("/users")
async def create_user(request):
data = await request.json()
return {"message": "User created", "data": data}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)Access the API documentation:
- Swagger UI:
http://localhost:8000/docs - OpenAPI JSON:
http://localhost:8000/openapi.json
Write synchronous code, the framework automatically handles it:
@app.get("/users/{id}")
def get_user(id: int): # Regular function!
# Blocking operations automatically executed in executor
user = database.get_user(id)
posts = database.get_user_posts(id)
return {"user": user, "posts": posts}Dependencies execute in parallel:
@app.get("/dashboard")
async def dashboard(
user: User = get_user(),
stats: Stats = get_stats(),
notifications: list = get_notifications()
):
# All three functions execute in parallel!
return {
"user": user,
"stats": stats,
"notifications": notifications
}React to events in your application:
@app.react("order:created")
async def on_order_created(event):
order = event.data
await inventory.reserve(order.items)
await payment.process(order)
await shipping.schedule(order)
# Emit event
await app.emit("order:created", order_data)Compose functions into pipelines:
@app.pipeline([
authenticate,
authorize,
validate,
transform,
save
])
def create_resource(data: ResourceData):
return {"id": data.id, "status": "created"}Conditional routing based on conditions:
@app.when(lambda req: req.headers.get("X-Client") == "mobile")
def mobile_handler(request):
return {"mobile": True}
@app.when(lambda req: req.path.startswith("/api/v2"))
def v2_handler(request):
return {"version": "2.0"}OpenAPI/Swagger documentation is automatically generated:
app = QakeAPI(
title="My API",
version="1.2.0",
description="API documentation"
)
# All routes are automatically documented
@app.get("/users/{id}")
def get_user(id: int):
"""Get user by ID."""
return {"id": id}Real-time communication:
@app.websocket("/ws/{room}")
async def websocket_handler(websocket: WebSocket, room: str):
await websocket.accept()
await websocket.send_json({"message": f"Welcome to {room}!"})
async for message in websocket.iter_json():
await websocket.send_json({"echo": message})Run tasks asynchronously:
from qakeapi.core.background import add_background_task
@app.post("/process")
async def process_data(request):
data = await request.json()
# Run task in background
await add_background_task(process_heavy_task, data)
return {"message": "Processing started"}Customize request/response processing:
from qakeapi import CORSMiddleware, LoggingMiddleware, RequestSizeLimitMiddleware
app.add_middleware(CORSMiddleware(allow_origins=["*"]))
app.add_middleware(LoggingMiddleware())
app.add_middleware(RequestSizeLimitMiddleware(max_size=10 * 1024 * 1024)) # 10MBClean architecture with dependency injection:
from qakeapi import QakeAPI, Depends
app = QakeAPI()
def get_database():
return Database()
@app.get("/users")
async def get_users(db = Depends(get_database)):
return await db.get_users()Protect your API with rate limiting:
from qakeapi import rate_limit
@rate_limit(requests_per_minute=60)
@app.get("/api/data")
def get_data():
return {"data": "..."}Cache responses for better performance:
from qakeapi import cache
@cache(ttl=300) # Cache for 5 minutes
@app.get("/expensive-operation")
def expensive_operation():
return {"result": compute_expensive_result()}Handle file uploads with validation and security:
from qakeapi import QakeAPI, FileUpload, IMAGE_TYPES
@app.post("/upload")
async def upload_image(file: FileUpload):
# Validate file type
if not file.validate_type(IMAGE_TYPES):
return {"error": "Only images"}, 400
# Validate size (5MB)
if not file.validate_size(5 * 1024 * 1024):
return {"error": "File too large"}, 400
# Save file
path = await file.save("uploads/")
return {"path": path}qakeapi/
βββ core/ # Core components
β βββ app.py # Main QakeAPI class
β βββ hybrid.py # Hybrid executor (syncβasync)
β βββ router.py # Smart router (Trie-optimized)
β βββ reactive.py # Reactive engine
β βββ parallel.py # Parallel resolver
β βββ pipeline.py # Pipeline processor
β βββ request.py # HTTP Request
β βββ response.py # HTTP Response
β βββ middleware.py # Middleware system
β βββ websocket.py # WebSocket support
β βββ background.py # Background tasks
β βββ openapi.py # OpenAPI generation
β βββ files.py # File upload handling
β βββ dependencies.py # Dependency Injection
β βββ validation.py # Data validation
β βββ rate_limit.py # Rate limiting
β βββ caching.py # Response caching
β βββ logging.py # Logging system
β βββ exceptions.py # HTTP exceptions
βββ utils/ # Utilities
Full documentation is available in the docs/ directory:
- Getting Started - Quick start guide
- File Upload - File upload handling
- Routing Guide - Routing, handlers, and performance optimizations
- Dependency Injection - DI system for clean architecture
- Reactive System - Event-driven architecture
- Parallel Dependencies - Parallel dependency resolution
- Pipelines - Function pipelines
- Middleware - Middleware system and security
- WebSocket - WebSocket support
- Background Tasks - Background processing
- OpenAPI - API documentation
- API Reference - Complete API reference
Check out the examples/ directory for complete examples:
basic_example.py- Basic features demonstrationcomplete_example.py- Full feature showcasefile_upload_example.py- File upload handlingfinancial_calculator.py- Complex real-world application
- Python 3.9+
- uvicorn (optional, for running the server)
MIT License - see LICENSE for details.
QakeAPI is built from scratch using only Python standard library, demonstrating a new approach to web frameworks.
QakeAPI - Build modern APIs with revolutionary approach! π
Made with β€οΈ by the QakeAPI team