🚀 Overview

VayuAPI is a lightning-fast, production-ready Python web framework built for high-performance APIs with native async/await support, advanced concurrency primitives, and seamless Django ORM integration.

Key Highlights

Design Philosophy

• Performance First: Every decision prioritizes sub-microsecond latency
• Developer Experience: Minimal boilerplate, maximum productivity
• Security by Default: Built-in protections for common vulnerabilities
• Production Ready: Deploy to 145K+ RPS with confidence
• Scalability: Native support for horizontal scaling

💡 Why VayuAPI?

VayuAPI solves problems that existing frameworks don't address:

Complete Solution Example

Here's how VayuAPI brings it all together:

from vayuapi import VayuAPI, run_in_thread, Semaphore
from vayuapi.security import JWTAuth, AESEncryption
from vayuapi.cache import AsyncLRUCache
from django.contrib.auth.models import User

app = VayuAPI(
    title="My API",
    admin_enabled=True,
    cors_enabled=True
)

# Security
jwt = JWTAuth(secret_key="your-secret-key")
encryption = AESEncryption(key="encryption-key")

# Concurrency
db_sem = Semaphore(10)  # Limit DB connections
cache = AsyncLRUCache(max_size=1000, ttl=300)

# Dependency: Get current user from JWT
async def get_current_user(token: str = Header(...)):
    return await jwt.verify_token(token)

# Endpoint: Protected, cached, with resource limiting
@app.get("/user/{id}")
@cache.cached
async def get_user(
    id: int,
    current_user: User = Depends(get_current_user)
):
    async with db_sem:  # Limit concurrent DB queries
        return await User.objects.filter(id=id).afirst()

# Endpoint: CPU-intensive task
@app.post("/process")
async def process_data(data: list):
    result = await run_in_thread(expensive_computation, data)
    return {"result": result}

# Admin panel auto-generated at /admin
if app.admin_enabled:
    app.admin_panel.register(User)

⚡ Quick Start

5-Minute Guide

# Install
pip install vayuapi

# Create app.py
from vayuapi import VayuAPI
from pydantic import BaseModel

app = VayuAPI()

class Item(BaseModel):
    name: str
    price: float

@app.get("/")
async def root():
    return {"message": "Hello, World!"}

@app.post("/items")
async def create_item(item: Item):
    return item

# Run
if __name__ == "__main__":
    app.run()

2-Minute to Production

from vayuapi import VayuAPI
from pydantic import BaseModel
import os

app = VayuAPI(
    title="Production API",
    debug=False,
    admin_enabled=True
)

# Your code here...

if __name__ == "__main__":
    port = int(os.getenv("PORT", 8000))
    app.run(
        host="0.0.0.0",
        port=port,
        workers=4  # Production workers
    )

📦 Installation

Basic Installation

pip install vayuapi

With All Features

pip install vayuapi[all]

Feature-Specific

pip install vayuapi[django]      # Django ORM
pip install vayuapi[orm]         # Tortoise ORM
pip install vayuapi[ai,rag]      # AI/ML features
pip install vayuapi[security]    # Security features
pip install vayuapi[scheduler]   # Task scheduling
pip install vayuapi[vector]      # Vector databases

System Requirements

• Python 3.12 - 3.16
• pip (Python package manager)
• 2GB RAM minimum (4GB recommended)
• Linux, macOS, or Windows

🎯 Core Features

Routing

@app.get("/items")
async def list_items():
    return {"items": []}

@app.post("/items")
async def create_item(item: Item):
    return item

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"deleted": item_id}

# Query parameters
@app.get("/search")
async def search(q: str, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

Response Types

from vayuapi import JSONResponse, HTMLResponse, FileResponse
from vayuapi.responses import StreamingResponse

# JSON (default)
@app.get("/json")
async def json_response():
    return {"data": "value"}

# HTML
@app.get("/html")
async def html_response():
    return HTMLResponse(content="
Hello
")

# File Download
@app.get("/download")
async def download():
    return FileResponse("path/to/file.pdf")

# Streaming
@app.get("/stream")
async def stream():
    async def generate():
        for i in range(100):
            yield f"data: {i}\n"
    return StreamingResponse(generate())

Request Validation

from pydantic import BaseModel, Field, validator

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    price: float = Field(..., gt=0)
    description: str = Field(default="", max_length=500)

    @validator('price')
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('Price must be positive')
        return v

@app.post("/items")
async def create_item(item: Item):
    return item

Dependency Injection

from vayuapi import Depends

def get_db():
    db = Database()
    return db

async def get_current_user(token: str = Header(...)):
    user = await verify_token(token)
    return user

@app.get("/items")
async def list_items(db = Depends(get_db)):
    return await db.query("SELECT * FROM items")

@app.get("/profile")
async def profile(user = Depends(get_current_user)):
    return user

WebSocket Support

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except ConnectionClosedOK:
        pass

Middleware

from vayuapi import Middleware
from starlette.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware
@app.middleware("http")
async def add_process_time_header(request, call_next):
    import time
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Exception Handling

from vayuapi import HTTPException

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id == 0:
        raise HTTPException(status_code=400, detail="Invalid ID")
    if item_id > 1000:
        raise HTTPException(status_code=404, detail="Not found")
    return {"item_id": item_id}

@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        {"error": str(exc)},
        status_code=400
    )

Event Handlers

@app.on_event("startup")
async def startup_event():
    print("Application startup")
    # Initialize resources

@app.on_event("shutdown")
async def shutdown_event():
    print("Application shutdown")
    # Cleanup resources

Background Tasks

from vayuapi import BackgroundTasks

def write_notification(email: str, message: str):
    print(f"Sending email to {email}: {message}")

@app.post("/send-notification/")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="Email body")
    return {"message": "Notification sent in background"}

File Uploads

from vayuapi import UploadFile

@app.post("/upload")
async def upload(file: UploadFile):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents)
    }

@app.post("/upload-multiple")
async def upload_multiple(files: list[UploadFile]):
    results = []
    for file in files:
        contents = await file.read()
        results.append({
            "filename": file.filename,
            "size": len(contents)
        })
    return results

🔥 Advanced Features

Native Concurrency

from vayuapi import run_in_thread, run_in_process, Semaphore, RateLimiter

# Thread Pool - for blocking I/O
def blocking_io():
    import time
    time.sleep(1)
    return "done"

@app.get("/thread")
async def use_thread():
    result = await run_in_thread(blocking_io)
    return result

# Process Pool - for CPU-intensive tasks
def cpu_intensive(data):
    return sum([x**2 for x in data])

@app.post("/process")
async def cpu_work(data: list):
    result = await run_in_process(cpu_intensive, data)
    return result

# Semaphore - limit concurrent operations
db_sem = Semaphore(10)

@app.get("/db")
async def db_query():
    async with db_sem:
        return await database.query()

# Rate Limiter
limiter = RateLimiter(rate=100, per=60)

@app.get("/api")
async def rate_limited(request):
    if not await limiter.check(request.client.host):
        raise HTTPException(429, "Rate limit exceeded")
    return {"data": "value"}

Database Integration

# Django ORM (async)
from django.contrib.auth.models import User
from asgiref.sync import sync_to_async

@app.get("/users")
async def list_users():
    @sync_to_async
    def get_users():
        return list(User.objects.all())
    return await get_users()

# Tortoise ORM (native async)
from tortoise import fields
from tortoise.models import Model

class User(Model):
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=100, unique=True)

@app.get("/users")
async def get_users():
    users = await User.all()
    return users

# SQLAlchemy (async)
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:password@localhost/db")

@app.get("/users")
async def get_users():
    async with engine.begin() as conn:
        result = await conn.execute("SELECT * FROM users")
        return result.fetchall()

Admin Panel

app = VayuAPI(admin_enabled=True, admin_path="/admin")

# Auto-generated admin for Django models
from django.contrib.auth.models import User

if app.admin_panel:
    app.admin_panel.register(User)

# Access at http://localhost:8000/admin

JWT Authentication

from vayuapi.security.jwt import JWTHandler

jwt_handler = JWTHandler(secret_key="your-secret-key")

# Generate token
@app.post("/login")
async def login(username: str, password: str):
    if verify_password(username, password):
        token = jwt_handler.create_access_token(
            data={"sub": username},
            expires_in=3600
        )
        return {"access_token": token}
    raise HTTPException(401, "Invalid credentials")

# Verify token
async def get_current_user(token: str = Header(...)):
    payload = await jwt_handler.verify_token(token)
    return payload

@app.get("/protected")
async def protected(user = Depends(get_current_user)):
    return {"message": f"Hello {user['sub']}"}

Security

from vayuapi.security.encryption import AESEncryption, RSAEncryption, HashingUtility

# Password Hashing
hasher = HashingUtility()
hashed = hasher.hash_password("password123")
is_valid = hasher.verify_password("password123", hashed)

# AES Encryption
aes = AESEncryption(key="32-char-key-minimum-required")
encrypted = aes.encrypt("sensitive data")
decrypted = aes.decrypt(encrypted)

# RSA Encryption
rsa = RSAEncryption()
public_key = rsa.generate_keys()
encrypted = rsa.encrypt("data", public_key)
decrypted = rsa.decrypt(encrypted)

Task Scheduling

from vayuapi.scheduler import Scheduler

scheduler = Scheduler()

# Cron-style
@scheduler.cron("0 0 * * *")  # Daily at midnight
async def daily_task():
    print("Running daily task")

# Interval-based
@scheduler.interval(seconds=300)  # Every 5 minutes
async def periodic_task():
    print("Running periodic task")

# One-time
@scheduler.once(delay=60)  # After 60 seconds
async def delayed_task():
    print("Running one-time task")

AI/ML Integration

from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

llm = ChatOpenAI(api_key="your-key")
memory = ConversationBufferMemory()
chain = ConversationChain(llm=llm, memory=memory)

@app.post("/chat")
async def chat(message: str):
    response = await chain.arun(input=message)
    return {"response": response}

# RAG Pipeline
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone

loader = TextLoader("documents.txt")
documents = loader.load()
splitter = CharacterTextSplitter(chunk_size=1000)
chunks = splitter.split_documents(documents)

embeddings = OpenAIEmbeddings()
vectorstore = Pinecone.from_documents(chunks, embeddings)

@app.post("/ask")
async def ask(question: str):
    docs = await vectorstore.asimilarity_search(question)
    return {"docs": docs}

Router System

from vayuapi import Router

router = Router(prefix="/api/v1")

@router.get("/items")
async def list_items():
    return {"items": []}

@router.post("/items")
async def create_item(item: Item):
    return item

app.include_router(router)

OpenAPI & Documentation

@app.get(
    "/items",
    tags=["items"],
    summary="List all items",
    description="Get a complete list of all items",
    response_description="List of items"
)
async def list_items():
    """
    Get all items from the database.
    
    Returns:
        - A list of items
    """
    return {"items": []}

Status Codes

from vayuapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

@app.delete("/items/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(id: int):
    return None

📊 Framework Comparison

VayuAPI vs FastAPI

VayuAPI vs Django

VayuAPI vs Flask

VayuAPI vs Sanic

⚡ Performance Benchmarks

Simple JSON Response

Database Query

CPU-Intensive Task

✨ Complete Feature List

💼 Use Cases

RESTful APIs

from vayuapi import VayuAPI
from pydantic import BaseModel

app = VayuAPI(title="Blog API")

class Post(BaseModel):
    title: str
    content: str
    author: str

@app.get("/posts")
async def list_posts():
    return await Post.objects.all()

@app.post("/posts")
async def create_post(post: Post):
    return await Post.objects.create(**post.dict())

@app.get("/posts/{id}")
async def get_post(id: int):
    return await Post.objects.get(id=id)

Microservices

# Service 1: User Service
@app.get("/users/{id}")
async def get_user(id: int):
    return await User.objects.get(id=id)

# Service 2: Order Service with inter-service communication
@app.get("/orders/{id}")
async def get_order(id: int):
    async with http_pool.acquire() as client:
        user = await client.get(f"http://user-service/users/{order.user_id}")
    return {"order": order, "user": user.json()}

Real-Time Applications

connections = {}

@app.websocket("/ws/chat/{room}")
async def chat_room(websocket: WebSocket, room: str):
    await websocket.accept()
    connections[room] = connections.get(room, []) + [websocket]

    try:
        while True:
            message = await websocket.receive_json()
            for conn in connections[room]:
                await conn.send_json(message)
    finally:
        connections[room].remove(websocket)

Background Processing

@app.post("/process-video")
async def process_video(video_url: str):
    tasks = BackgroundTasks()
    tasks.add(download_video, video_url)
    tasks.add(convert_format, video_url)
    tasks.add(upload_to_s3, video_url)
    asyncio.create_task(tasks.execute())
    return {"status": "processing"}

Machine Learning APIs

model = tf.keras.models.load_model("model.h5")
cache = AsyncLRUCache(max_size=10000, ttl=3600)

@app.post("/predict")
async def predict(data: list):
    result = await run_in_process(model.predict, data)
    return {"prediction": result.tolist()}

Admin Dashboards

app = VayuAPI(
    title="Admin Dashboard",
    admin_enabled=True,
    admin_path="/admin"
)

# Models automatically get admin interface
if app.admin_panel:
    app.admin_panel.register(User)
    app.admin_panel.register(Product)

Data Processing Pipelines

batcher = BatchProcessor(
    process_func=insert_batch,
    max_batch_size=1000,
    max_wait_time=0.1
)

@app.post("/ingest")
async def ingest_data(record: dict):
    await batcher.add(record)
    return {"status": "queued"}

# 1000 requests → 1 database call

API Gateway

services = {
    "users": "http://user-service:8001",
    "orders": "http://order-service:8002"
}

@app.api_route("/{service}/{path:path}", methods=["GET", "POST"])
async def gateway(request, service: str, path: str):
    target_url = f"{services[service]}/{path}"
    async with http_pool.acquire() as client:
        return await client.request(
            method=request.method,
            url=target_url
        )

AI Agents & Chatbots

from pydantic_ai import Agent

agent = Agent(model="gpt-4-turbo")

@app.post("/chat")
async def chat(message: str, user_id: str):
    # User context stored in database
    user_context = await get_user_context(user_id)
    
    # Agent processes with full context
    response = await agent.run(
        message,
        context=user_context
    )
    
    # Store interaction for RAG
    await store_interaction(user_id, message, response)
    
    return {"response": response}

🎯 Best Practices

Performance Optimization

# ✅ DO: Use thread pool for blocking operations
@app.get("/blocking")
async def blocking_op():
    result = await run_in_thread(blocking_function)
    return result

# ❌ DON'T: Block the event loop
@app.get("/bad")
async def bad():
    time.sleep(1)  # Blocks everything!
    return {}

# ✅ DO: Use semaphores to limit concurrent operations
db_sem = Semaphore(10)

@app.get("/db")
async def query():
    async with db_sem:
        return await db.query()

# ✅ DO: Cache expensive operations
cache = AsyncLRUCache(max_size=1000, ttl=300)

@cache.cached
async def expensive():
    return await slow_operation()

Error Handling

# ✅ DO: Use specific exceptions
@app.get("/item/{id}")
async def get_item(id: int):
    item = await Item.objects.filter(id=id).first()
    if not item:
        raise NotFoundException(f"Item {id} not found")
    return item

# ✅ DO: Log errors properly
import logging
logger = logging.getLogger(__name__)

@app.get("/safe")
async def safe_endpoint():
    try:
        result = await risky_operation()
        return result
    except Exception as e:
        logger.error(f"Error: {e}", exc_info=True)
        raise InternalServerErrorException()

Security

# ✅ DO: Validate all inputs
from pydantic import BaseModel, EmailStr, constr

class UserInput(BaseModel):
    username: constr(regex="^[a-zA-Z0-9_]+$", min_length=3)
    email: EmailStr

# ✅ DO: Use environment variables for secrets
import os
SECRET_KEY = os.getenv("SECRET_KEY")

# ✅ DO: Use HTTPS in production
app.run(
    ssl_keyfile="key.pem",
    ssl_certfile="cert.pem"
)

# ✅ DO: Implement rate limiting
limiter = RateLimiter(rate=100, per=60)

@app.get("/api")
async def api(request):
    if not await limiter.check(request.client.host):
        raise HTTPException(429, "Rate limit exceeded")
    return {"data": "..."}

Database Queries

# ✅ DO: Use connection pooling
db_pool = ConnectionPool(
    create_func=create_db_connection,
    max_size=20
)

# ✅ DO: Use batch operations
batcher = BatchProcessor(fetch_users_batch, max_batch_size=50)

@app.get("/user/{id}")
async def get_user(id: int):
    return await batcher.add(id)

# ✅ DO: Use select_related to avoid N+1 queries
@app.get("/posts")
async def good_posts():
    posts = await Post.objects.select_related('author').all()
    return posts

Testing

from vayuapi.testing import TestClient

client = TestClient(app)

def test_create_user():
    response = client.post(
        "/users",
        json={"username": "test", "email": "test@example.com"}
    )
    assert response.status_code == 201

def test_authentication():
    response = client.post(
        "/login",
        json={"username": "test", "password": "password"}
    )
    token = response.json()["access_token"]

    response = client.get(
        "/protected",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200

🔄 Migration Guides

From FastAPI to VayuAPI

VayuAPI is designed to be compatible with FastAPI. Most code works as-is:

# FastAPI code (works in VayuAPI!)
from vayuapi import VayuAPI, Depends

app = VayuAPI()

@app.get("/items")
async def list_items():
    return {"items": []}

# Additional VayuAPI features:
# - Built-in admin panel
# - Native thread/process pools
# - Built-in JWT auth
# - Async caching
# - Semaphores for resource limiting

From Flask to VayuAPI

# Flask code
@app.route("/users", methods=["GET", "POST"])
def users():
    if request.method == "GET":
        return jsonify({"users": []})
    return jsonify({"created": data}), 201

# VayuAPI equivalent
@app.get("/users")
async def list_users():
    return {"users": []}

@app.post("/users")
async def create_user(data: dict):
    return JSONResponse({"created": data}, status_code=201)

From Django REST Framework to VayuAPI

# Django REST Framework
class UserList(APIView):
    def get(self, request):
        users = User.objects.all()
        serializer = UserSerializer(users, many=True)
        return Response(serializer.data)

# VayuAPI equivalent (much faster!)
@app.get("/users")
async def list_users():
    @sync_to_async
    def get_users():
        return list(User.objects.all())
    users = await get_users()
    return {"users": [user.username for user in users]}

📚 API Reference

Core Classes

VayuAPI Class

class VayuAPI:
    def __init__(
        self,
        title: str = "VayuAPI",
        version: str = "0.1.0",
        description: str = "",
        debug: bool = False,
        docs_enabled: bool = True,
        admin_enabled: bool = False,
        cors_enabled: bool = True,
        ...
    )

    # HTTP Methods
    def get(self, path: str, **kwargs) -> Callable
    def post(self, path: str, **kwargs) -> Callable
    def put(self, path: str, **kwargs) -> Callable
    def delete(self, path: str, **kwargs) -> Callable
    def patch(self, path: str, **kwargs) -> Callable
    
    # WebSocket & Events
    def websocket(self, path: str) -> Callable
    def on_event(self, event_type: str) -> Callable
    
    # Middleware & Routing
    def middleware(self, middleware_type: str) -> Callable
    def add_middleware(self, middleware_class, **options)
    def include_router(self, router: Router)
    
    # Execution
    def run(self, host: str = "127.0.0.1", port: int = 8000, **kwargs)

Concurrency Utilities

async def run_in_thread(func: Callable, *args, **kwargs) -> Any
async def run_in_process(func: Callable, *args, **kwargs) -> Any

class Semaphore:
    def __init__(self, value: int = 1)
    async def __aenter__(self)
    async def __aexit__(self, exc_type, exc_val, exc_tb)

class RateLimiter:
    def __init__(self, rate: int, per: float)
    async def check(self, key: str) -> bool
    async def reset(self, key: str)

class AsyncLRUCache:
    def __init__(self, max_size: int = 128, ttl: Optional[float] = None)
    def cached(self, func: Callable) -> Callable
    async def clear(self)

Security Classes

class JWTHandler:
    def __init__(self, secret_key: str)
    def create_access_token(self, data: dict, expires_in: int) -> str
    async def verify_token(self, token: str) -> dict

class AESEncryption:
    def __init__(self, key: str)
    def encrypt(self, data: str) -> str
    def decrypt(self, encrypted: str) -> str

class HashingUtility:
    def hash_password(self, password: str) -> str
    def verify_password(self, password: str, hashed: str) -> bool

Status Codes

from vayuapi import status

status.HTTP_200_OK
status.HTTP_201_CREATED
status.HTTP_204_NO_CONTENT
status.HTTP_400_BAD_REQUEST
status.HTTP_401_UNAUTHORIZED
status.HTTP_403_FORBIDDEN
status.HTTP_404_NOT_FOUND
status.HTTP_500_INTERNAL_SERVER_ERROR

# Helpers
status.is_success(200)        # True
status.is_error(404)          # True
status.get_reason_phrase(404) # "Not Found"

🎨 Templating, Static Files, and Mounting

VayuAPI includes full support for Jinja2 templating, static file serving, and mounting sub-applications, making it a complete web framework compatible with FastAPI's features.

Jinja2 Templating

Installation

Jinja2 is included in the base requirements:

pip install jinja2

Basic Usage

from vayuapi import VayuAPI, Request
from vayuapi import Jinja2Templates

app = VayuAPI()

# Initialize templates directory
templates = Jinja2Templates(directory="templates")

@app.get("/")
async def home(request: Request):
    return templates.TemplateResponse(
        "home.html",
        {"request": request, "title": "Home", "message": "Hello!"}
    )

Template Inheritance

templates/base.html:

<!DOCTYPE html>
<html>
<head>
    <title>{% block title %}My App{% endblock %}</title>
    <link rel="stylesheet" href="/static/css/style.css">
</head>
<body>
    <nav>
        <a href="/">Home</a>
        <a href="/about">About</a>
    </nav>

    <main>
        {% block content %}{% endblock %}
    </main>

    <footer>
        &copy; 2026 My App
    </footer>
</body>
</html>

templates/home.html:

{% extends "base.html" %}

{% block title %}{{ title }}{% endblock %}

{% block content %}
<h1>{{ message }}</h1>
<ul>
{% for item in items %}
    <li>{{ item }}</li>
{% endfor %}
</ul>
{% endblock %}

Advanced Features

from vayuapi import Jinja2Templates

# Custom Jinja2 environment options
templates = Jinja2Templates(
    directory="templates",
    autoescape=True,
    extensions=['jinja2.ext.debug']
)

# Render from string
html = templates.render_string(
    "Hello {{ name }}!",
    {"name": "World"}
)

# Get template object
template = templates.get_template("custom.html")
content = template.render({"data": "value"})

Static Files

Basic Usage

from vayuapi import VayuAPI
from vayuapi import StaticFiles

app = VayuAPI()

# Mount static files
app.mount("/static", StaticFiles(directory="static"), name="static")

Directory Structure

project/
├── main.py
├── static/
│   ├── css/
│   │   └── style.css
│   ├── js/
│   │   └── app.js
│   └── images/
│       └── logo.png
└── templates/
    └── index.html

Accessing Static Files

In your templates:

<link rel="stylesheet" href="/static/css/style.css">
<script src="/static/js/app.js"></script>
<img src="/static/images/logo.png" alt="Logo">

Advanced Configuration

from vayuapi import StaticFiles

# Serve with directory indexes
app.mount(
    "/static",
    StaticFiles(directory="static", html=True),
    name="static"
)

# Multiple static directories from packages
app.mount(
    "/assets",
    StaticFiles(packages=["mypackage"]),
    name="assets"
)

# Follow symbolic links
app.mount(
    "/files",
    StaticFiles(directory="files", follow_symlink=True),
    name="files"
)

Mounting Sub-Applications

Basic Mounting

from vayuapi import VayuAPI

# Main application
app = VayuAPI(title="Main App")

# Sub-application
admin_app = VayuAPI(title="Admin Panel")

@admin_app.get("/")
async def admin_home():
    return {"message": "Admin Home"}

@admin_app.get("/users")
async def admin_users():
    return {"users": []}

# Mount the admin app
app.mount("/admin", admin_app)

Now accessible at:

• http://localhost:8000/admin/
• http://localhost:8000/admin/users

Mount Multiple Applications

from vayuapi import VayuAPI, StaticFiles

app = VayuAPI()

# Mount static files
app.mount("/static", StaticFiles(directory="static"), name="static")

# Mount API v1
api_v1 = VayuAPI(title="API v1")
@api_v1.get("/users")
async def v1_users():
    return {"version": "1.0", "users": []}

app.mount("/api/v1", api_v1)

# Mount API v2
api_v2 = VayuAPI(title="API v2")
@api_v2.get("/users")
async def v2_users():
    return {"version": "2.0", "users": []}

app.mount("/api/v2", api_v2)

# Mount admin panel
admin = VayuAPI(title="Admin")
@admin.get("/")
async def admin_home():
    return {"message": "Admin Panel"}

app.mount("/admin", admin)

Complete Example

from vayuapi import VayuAPI, Request, Jinja2Templates, StaticFiles

# Main application
app = VayuAPI(
    title="My Web App",
    version="1.0.0",
    docs_enabled=True
)

# Set up templates and static files
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")

# ========================================
# Template Routes
# ========================================

@app.get("/")
async def home(request: Request):
    return templates.TemplateResponse(
        "home.html",
        {
            "request": request,
            "title": "Home",
            "items": ["Feature 1", "Feature 2", "Feature 3"]
        }
    )

@app.get("/about")
async def about(request: Request):
    return templates.TemplateResponse(
        "about.html",
        {"request": request, "title": "About"}
    )

@app.get("/users/{user_id}")
async def user_profile(request: Request, user_id: int):
    user = {"id": user_id, "name": f"User {user_id}"}
    return templates.TemplateResponse(
        "user.html",
        {"request": request, "user": user}
    )

# ========================================
# API Routes
# ========================================

@app.get("/api/health")
async def health():
    return {"status": "healthy"}

@app.get("/api/users")
async def list_users():
    return {"users": [{"id": 1, "name": "Alice"}]}

# ========================================
# Mount Sub-Applications
# ========================================

# API v1
api_v1 = VayuAPI(title="API v1")

@api_v1.get("/posts")
async def v1_posts():
    return {"version": "1.0", "posts": []}

app.mount("/api/v1", api_v1)

# Admin panel
admin_app = VayuAPI(title="Admin")

@admin_app.get("/")
async def admin_home():
    return {"message": "Admin Panel"}

@admin_app.get("/stats")
async def admin_stats():
    return {"users": 100, "posts": 500}

app.mount("/admin", admin_app)

# ========================================
# Run Application
# ========================================

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, reload=True)

Best Practices

project/
├── main.py
├── templates/
│   ├── base.html
│   ├── home.html
│   ├── about.html
│   └── components/
│       ├── header.html
│       └── footer.html
├── static/
│   ├── css/
│   │   └── style.css
│   ├── js/
│   │   └── app.js
│   └── images/
│       └── logo.png
└── apps/
    ├── api/
    │   └── routes.py
    └── admin/
        └── panel.py

Error Handling

from vayuapi import HTTPException, Request
from vayuapi import Jinja2Templates

templates = Jinja2Templates(directory="templates")

@app.exception_handler(404)
async def not_found(request: Request, exc):
    return templates.TemplateResponse(
        "404.html",
        {"request": request},
        status_code=404
    )

@app.exception_handler(500)
async def server_error(request: Request, exc):
    return templates.TemplateResponse(
        "500.html",
        {"request": request, "error": str(exc)},
        status_code=500
    )

🚀 Production Deployment

Uvicorn (Single Worker)

# Development
uvicorn main:app --reload

# Production
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

Gunicorn + Uvicorn (Recommended)

gunicorn main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 30 \
    --graceful-timeout 15 \
    --keep-alive 5

Docker

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["gunicorn", "main:app", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000"]

Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: vayuapi
spec:
  replicas: 3
  selector:
    matchLabels:
      app: vayuapi
  template:
    metadata:
      labels:
        app: vayuapi
    spec:
      containers:
      - name: vayuapi
        image: your-registry/vayuapi:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "2000m"
            memory: "2Gi"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000

Nginx (Reverse Proxy)

upstream vayuapi {
    server 127.0.0.1:8000;
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
    server 127.0.0.1:8003;
    keepalive 32;
}

server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://vayuapi;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
    }
}

Environment Variables

# Application
export APP_ENV=production
export DEBUG=false
export LOG_LEVEL=info

# Database
export DATABASE_URL=postgresql://localhost/mydb
export DB_POOL_SIZE=20

# Security
export SECRET_KEY=your-secret-key
export JWT_SECRET=your-jwt-secret

# External Services
export OPENAI_API_KEY=your-api-key
export REDIS_URL=redis://localhost:6379

🤝 Community & Support

Documentation

• Complete Guide: https://vayuapi.amrits.in
• API Reference: https://vayuapi.amrits.in/api-reference
• Examples: GitHub Examples

Getting Help

• GitHub Discussions: https://github.com/vayuapi/vayuapi/discussions
• Discord Community: https://discord.gg/vayuapi
• Stack Overflow: Tag vayuapi
• Email Support:codeswithalok@gmail.com

Contributing

• Report Bugs: GitHub Issues
• Feature Requests: GitHub Discussions
• Pull Requests: GitHub

License

VayuAPI is released under the MIT License. See LICENSE for details.

Acknowledgments

VayuAPI is built on top of these excellent projects:

• Starlette - ASGI framework
• Pydantic - Data validation
• Uvicorn - ASGI server
• Django - ORM integration
• Tortoise ORM - Async ORM

Quick Links

• 🏠 Website
• 📘 Documentation
• 💻 GitHub
• 📦 PyPI
• 💬 Discord
• X
• 📦 PyPi