VayuAPI Reference

Core

VayuAPI

Main application class for creating an API server.

from vayuapi import VayuAPI

app = VayuAPI(
    title="My API",
    version="1.0.0",
    description="API description",
    debug=False,
    admin_enabled=True,
    admin_path="/admin",
    cors_enabled=True,
    allowed_origins=["*"],
    middleware=[],
    lifespan_handlers=[]
)

Routing

HTTP Methods

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

@app.post("/users")
async def create_user(user: User):
    return {"user": user}

@app.put("/users/{id}")
async def update_user(id: int, user: User):
    return {"user": user}

@app.delete("/users/{id}")
async def delete_user(id: int):
    return {"deleted": id}

Path Parameters

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

@app.get("/posts/{slug}")
async def get_post(slug: str):
    return {"slug": slug}

Query Parameters

@app.get("/search")
async def search(q: str, limit: int = 10, offset: int = 0):
    return {
        "query": q,
        "limit": limit,
        "offset": offset
    }

Request & Response

Request Body with Pydantic

from pydantic import BaseModel

class User(BaseModel):
    name: str
    email: str
    age: int
    is_active: bool = True

@app.post("/users")
async def create_user(user: User):
    return {"user": user}

Response Models

from pydantic import BaseModel

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    return {"id": user_id, "name": "John", "email": "john@example.com"}

WebSocket

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

Middleware

Built-in Middleware

from vayuapi.middleware import CORSMiddleware, AuthMiddleware

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

Custom Middleware

from vayuapi import Middleware

class CustomMiddleware(Middleware):
    async def __call__(self, scope, receive, send):
        # Pre-processing
        await self.app(scope, receive, send)
        # Post-processing

ORM Integration

Tortoise ORM

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

app = VayuAPI()

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

    class Meta:
        table = "users"

Admin Panel

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

# Admin interface automatically generated
# Access at: http://localhost:8000/admin

Security

JWT Authentication

from vayuapi.security import create_access_token

@app.post("/login")
async def login(username: str, password: str):
    # Validate credentials
    token = create_access_token(data={"sub": username})
    return {"access_token": token, "token_type": "bearer"}

Encryption

from vayuapi.security import AESEncryption

cipher = AESEncryption()
encrypted = cipher.encrypt("sensitive data")
decrypted = cipher.decrypt(encrypted)

Utilities

Concurrency - Thread Pool

from vayuapi import run_in_thread, to_thread

# Function style
@app.get("/thread")
async def thread_example():
    result = await run_in_thread(blocking_function, arg1, arg2)
    return {"result": result}

# Decorator style
@to_thread
def blocking_file_read(filename: str):
    with open(filename, 'r') as f:
        return f.read()

@app.get("/read")
async def read_file():
    content = await blocking_file_read("data.txt")
    return {"content": content}

Concurrency - Process Pool

from vayuapi import run_in_process

@app.post("/compute")
async def compute(data: list):
    # True parallelism, bypasses GIL
    result = await run_in_process(cpu_intensive_task, data)
    return {"result": result}

Semaphores - Resource Limiting

from vayuapi import Semaphore

# Limit concurrent access
db_semaphore = Semaphore(10)
redis_semaphore = Semaphore(20)

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

Async LRU Cache

from vayuapi import AsyncLRUCache\n\ncache = AsyncLRUCache(max_size=1000, ttl=300)\n\n@cache.cached\nasync def expensive_operation(id: int):\n    return await slow_db_query(id)\n\n@app.get(\"/data/{id}\")\nasync def get_data(id: int):\n    # Cached for 5 minutes, 500x faster\n    return await expensive_operation(id)

\n\n

Dependency Injection

from vayuapi import Depends\n\nasync def get_current_user(token: str):
    # Validate and return user
    return {"id": 1, "username": "john"}

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

Status Codes

Code	Description
200	OK - Success
201	Created - Resource created
204	No Content - Success, no content
400	Bad Request - Invalid input
401	Unauthorized - Authentication required
403	Forbidden - Access denied
404	Not Found - Resource not found
500	Internal Server Error
502	Bad Gateway - Upstream server error
503	Service Unavailable - Server overloaded
429	Too Many Requests - Rate limited

Error Handling Best Practices

from vayuapi import HTTPException

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id < 1:
        raise HTTPException(status_code=400, detail="Invalid user ID")
    if not current_user.has_permission():
        raise HTTPException(status_code=403, detail="Access denied")
    user = await User.get_or_none(id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Request Logging

import logging

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request, call_next):
    logger.info(f"Request: {request.method} {request.url.path}")
    response = await call_next(request)
    logger.info(f"Response: {response.status_code}")
    return response

CORS Configuration

app = VayuAPI(
    cors_enabled=True,
    allowed_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"]
)

More Information

For detailed documentation, visit the Getting Started guide or check specific feature documentation.