VayuAPI Examples

Comprehensive examples demonstrating all VayuAPI features and capabilities.

Basic Examples

Beginner

1. Hello World

The simplest VayuAPI application:

from vayuapi import VayuAPI

app = VayuAPI()

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

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

Beginner

2. Basic Routing

HTTP methods and path parameters:

from vayuapi import VayuAPI
from pydantic import BaseModel

app = VayuAPI()

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

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

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

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

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

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

Beginner

3. Query Parameters

Filtering and pagination:

from vayuapi import VayuAPI, Query

app = VayuAPI()

@app.get("/search")
async def search(
    q: str = Query(..., min_length=3),
    page: int = Query(1, ge=1),
    limit: int = Query(10, ge=1, le=100)
):
    return {
        "query": q,
        "page": page,
        "limit": limit,
        "results": []
    }

@app.get("/filter")
async def filter_items(
    category: str = Query(None),
    min_price: float = Query(None),
    max_price: float = Query(None)
):
    return {"filters": {"category": category, "price": [min_price, max_price]}}

Intermediate Examples

Intermediate

4. Request Validation

Pydantic models with validation:

from pydantic import BaseModel, Field, EmailStr, validator

class CreateUser(BaseModel):
    username: str = Field(..., min_length=3, max_length=20)
    email: EmailStr
    age: int = Field(..., ge=18, le=120)
    bio: str = Field(None, max_length=500)

    @validator('username')
    def validate_username(cls, v):
        if v.lower() in ['admin', 'root']:
            raise ValueError('Username is reserved')
        return v

app = VayuAPI()

@app.post("/register")
async def register(user: CreateUser):
    # Validation happens automatically
    return {"user": user}

Intermediate

5. JWT Authentication

Secure token-based authentication:

from vayuapi import VayuAPI, Depends
from vayuapi.security import JWTHandler, JWTBearer
import os

app = VayuAPI()

jwt_handler = JWTHandler(
    secret_key=os.getenv("SECRET_KEY"),
    algorithm="HS256",
    access_token_expire_minutes=30
)

jwt_auth = JWTBearer(jwt_handler)

@app.post("/login")
async def login(username: str, password: str):
    if verify_credentials(username, password):
        token = jwt_handler.create_access_token(
            data={"sub": username, "user_id": 123}
        )
        return {"access_token": token, "token_type": "bearer"}

@app.get("/profile")
async def get_profile(payload = Depends(jwt_auth)):
    return {
        "user": payload.get("sub"),
        "user_id": payload.get("user_id")
    }

Intermediate

6. Dependency Injection

Reusable dependencies:

from vayuapi import VayuAPI, Depends

async def get_current_user(token: str = Header(...)):
    # Verify token and return user
    return {"user_id": 123, "username": "john"}

async def get_db():
    db = Database()
    try:
        yield db
    finally:
        await db.close()

app = VayuAPI()

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

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

Intermediate

7. WebSocket Communication

Real-time bidirectional communication:

from vayuapi import VayuAPI

app = VayuAPI()

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            # Broadcast to all connected clients
            await websocket.send_text(f"Echo: {data}")
    except Exception as e:
        await websocket.close(code=1000)

@app.websocket("/ws/notifications")
async def notifications(websocket):
    await websocket.accept()
    while True:
        # Send real-time notifications
        await websocket.send_json({
            "type": "notification",
            "message": "New event occurred"
        })
        await asyncio.sleep(5)

Intermediate

8. Database Integration (Tortoise ORM)

Async ORM for database operations:

from vayuapi import VayuAPI
from tortoise import fields
from tortoise.models import Model

app = VayuAPI(admin_enabled=True)

class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50, unique=True)
    email = fields.CharField(max_length=100)
    age = fields.IntField()
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "users"

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

@app.post("/users")
async def create_user(username: str, email: str, age: int):
    user = await User.create(username=username, email=email, age=age)
    return {"user": user}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await User.get(id=user_id)
    return {"user": user}

Intermediate

9. Middleware

Request/response processing:

from vayuapi import VayuAPI, Middleware

app = VayuAPI(
    cors_enabled=True,
    allowed_origins=["*"]
)

class LoggingMiddleware(Middleware):
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            method = scope["method"]
            path = scope["path"]
            print(f"Request: {method} {path}")
        await self.app(scope, receive, send)

class TimingMiddleware(Middleware):
    async def __call__(self, scope, receive, send):
        start_time = time.time()
        await self.app(scope, receive, send)
        duration = time.time() - start_time
        print(f"Request took {duration:.3f}s")

Advanced Examples

Advanced

10. Thread Pool for Blocking Operations

Offload CPU-bound tasks without blocking the event loop:

from vayuapi import VayuAPI, run_in_thread

app = VayuAPI()

def cpu_intensive_task(n: int):
    # Blocking CPU operation
    return sum(i*i for i in range(n))

@app.get("/compute/{n}")
async def compute(n: int):
    # Runs in thread pool (100+ RPS vs 0.2 RPS)
    result = await run_in_thread(cpu_intensive_task, n)
    return {"result": result}

@app.post("/process")
async def process_file(file_content: str):
    # File processing in thread pool
    processed = await run_in_thread(
        process_large_file,
        file_content
    )
    return {"processed": processed}

Advanced

11. Process Pool for CPU Parallelism

True parallelism bypassing Python's GIL:

from vayuapi import VayuAPI, run_in_process

app = VayuAPI()

def expensive_computation(data: list):
    # CPU-intensive operation
    return [x**2 for x in data]

@app.post("/compute/heavy")
async def heavy_compute(data: list):
    # Runs in separate process (full CPU core)
    result = await run_in_process(expensive_computation, data)
    return {"result": result}

@app.post("/machine-learning")
async def ml_predict(features: list):
    # ML model prediction in separate process
    prediction = await run_in_process(
        model.predict,
        features
    )
    return {"prediction": prediction}

Advanced

12. Semaphores for Resource Control

Prevent resource exhaustion with concurrent limits:

from vayuapi import VayuAPI, Semaphore

app = VayuAPI()

# Limit concurrent database connections
db_semaphore = Semaphore(10)

# Limit concurrent external API calls
api_semaphore = Semaphore(5)

@app.get("/query")
async def query_database():
    async with db_semaphore:
        # Max 10 concurrent queries
        result = await database.query("SELECT * FROM users")
        return {"result": result}

@app.get("/external-api")
async def call_external_api():
    async with api_semaphore:
        # Max 5 concurrent API calls
        response = await external_api.call()
        return {"data": response}

Advanced

13. Async LRU Cache

High-performance caching with automatic eviction:

from vayuapi import VayuAPI, AsyncLRUCache

app = VayuAPI()

cache = AsyncLRUCache(
    max_size=1000,  # Cache 1000 items max
    ttl=300  # 5 minutes TTL
)

@cache.cached
async def expensive_database_query(user_id: int):
    # Cache for 5 minutes
    return await database.query_user(user_id)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # First call hits DB, subsequent calls hit cache
    user = await expensive_database_query(user_id)
    return {"user": user}

@app.post("/invalidate")
async def invalidate_cache():
    cache.clear()
    return {"message": "Cache cleared"}

Advanced

14. Background Tasks

Execute tasks without blocking response:

from vayuapi import VayuAPI, BackgroundTasks

app = VayuAPI()

def send_email(email: str, subject: str):
    # Long-running task
    mail_service.send(email, subject)

def update_analytics(event: str):
    # Track event
    analytics.track(event)

@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
    # Register user
    user = await create_user(email)

    # Queue background tasks
    background_tasks.add_task(send_email, email, "Welcome")
    background_tasks.add_task(update_analytics, "user_registered")

    # Response sent immediately, tasks run in background
    return {"user": user}

@app.post("/order")
async def create_order(order_data: dict, background_tasks: BackgroundTasks):
    order = await database.create_order(order_data)
    
    # Queue multiple background tasks
    background_tasks.add_task(send_order_confirmation, order.id)
    background_tasks.add_task(notify_warehouse, order.id)
    background_tasks.add_task(update_inventory, order.items)
    
    return {"order": order}

Advanced

15. Rate Limiting

Prevent abuse with rate limiting:

from vayuapi import VayuAPI, RateLimiter

app = VayuAPI()

limiter = RateLimiter(rate=100, per=60)  # 100 req/min

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

# Per-endpoint rate limits
user_limiter = RateLimiter(rate=10, per=60)
admin_limiter = RateLimiter(rate=1000, per=60)

@app.post("/users/create")
async def create_user(request, user_data: dict):
    if not await user_limiter.check(request.client.host):
        return {"error": "Rate limit exceeded"}, 429
    
    return await create_user_in_db(user_data)

Advanced

16. Task Scheduling

Schedule periodic tasks:

from vayuapi import VayuAPI
from vayuapi.scheduler import scheduler

app = VayuAPI()

@scheduler.task(interval=60)  # Run every 60 seconds
async def cleanup_expired_sessions():
    await Session.delete_expired()
    print("Sessions cleaned up")

@scheduler.cron("0 0 * * *")  # Daily at midnight
async def generate_daily_report():
    report = await generate_report()
    await send_report_email(report)

@scheduler.cron("*/5 * * * *")  # Every 5 minutes
async def check_system_health():
    health = await check_health()
    if not health['status']:
        await alert_admin(health)

Advanced

17. Encryption & Security

Secure sensitive data:

from vayuapi import VayuAPI
from vayuapi.security import AESEncryption, HashingUtility
import os

app = VayuAPI()

aes = AESEncryption(key=os.getenv("ENCRYPTION_KEY"))
hasher = HashingUtility()

@app.post("/secure")
async def store_sensitive_data(data: str):
    # Encrypt data
    encrypted = aes.encrypt(data)
    
    # Store encrypted data
    await database.store(encrypted)
    
    return {"status": "stored"}

@app.post("/verify-password")
async def verify_password(password: str):
    # Hash password
    hashed = hasher.hash(password)
    
    # Store hashed password (never store plain text!)
    user = await User.create(
        username="john",
        password_hash=hashed
    )
    
    return {"user": user}

# Verify stored password
stored_hash = user.password_hash
if hasher.verify(input_password, stored_hash):
    return {"authenticated": True}

Advanced

18. Admin Panel

Auto-generated admin interface:

from vayuapi import VayuAPI
from tortoise import fields
from tortoise.models import Model

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

class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50)
    email = fields.CharField(max_length=100)
    is_active = fields.BooleanField(default=True)

class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    content = fields.TextField()
    author: fields.ForeignKeyRelation[User] = fields.ForeignKeyField(
        'models.User', related_name='posts'
    )

# Admin interface auto-generated at /admin
# Features:
# - List all records
# - Create new records
# - Edit existing records
# - Delete records
# - Search and filter

Advanced

19. Langchain Integration

AI/ML with Langchain:

from vayuapi import VayuAPI
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

app = VayuAPI()

llm = OpenAI(api_key="sk-...")
prompt = PromptTemplate(
    input_variables=["topic"],
    template="Write a summary about {topic}"
)
chain = LLMChain(llm=llm, prompt=prompt)

@app.post("/generate")
async def generate_content(topic: str):
    # Run LLM chain
    result = await chain.arun(topic=topic)
    return {"content": result}

@app.post("/chat")
async def chat(message: str):
    # Chat with AI
    response = await llm.agenerate([message])
    return {"response": response}

Advanced

20. Full Production App

Complete production-ready application:

from vayuapi import VayuAPI, Depends
from vayuapi.security import JWTHandler
from tortoise import fields
from tortoise.models import Model
import os

app = VayuAPI(
    title="Production API",
    version="1.0.0",
    admin_enabled=True,
    cors_enabled=True,
    allowed_origins=["https://yourdomain.com"]
)

# Models
class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50, unique=True)
    email = fields.CharField(max_length=100, unique=True)
    password_hash = fields.CharField(max_length=255)

# JWT
jwt_handler = JWTHandler(
    secret_key=os.getenv("SECRET_KEY"),
    access_token_expire_minutes=30
)

# Dependency
async def get_current_user(token: str = Header(...)):
    payload = jwt_handler.verify_token(token)
    user = await User.get(id=payload["user_id"])
    return user

# Routes
@app.post("/auth/login")
async def login(username: str, password: str):
    user = await User.get(username=username)
    if verify_password(password, user.password_hash):
        token = jwt_handler.create_access_token(
            data={"user_id": user.id}
        )
        return {"access_token": token}
    return {"error": "Invalid credentials"}, 401

@app.get("/users/me")
async def get_me(current_user = Depends(get_current_user)):
    return {"user": current_user}

@app.post("/users")
async def create_user(username: str, email: str, password: str):
    user = await User.create(
        username=username,
        email=email,
        password_hash=hash_password(password)
    )
    return {"user": user}