Server Application Factory

Overview

viveka-setu is the equivalent of Spring Boot's embedded server + @SpringBootApplication. It wraps FastAPI and Uvicorn so you focus on business logic, not server boilerplate.

Installation

With database support:

Quickstart

from viveka_server import VivekaApp, VivekaServer, VivekaAppConfig

app    = VivekaApp(VivekaAppConfig(name="my-service", version="1.0.0"))
server = VivekaServer(app)
server.run()

Your service starts at http://127.0.0.1:8000 with:

GET /        → {"service": "my-service", "version": "1.0.0", "status": "running"}
GET /health  → {"status": "healthy", "uptime_seconds": 3.2, ...}
GET /docs    → Swagger UI

Configuration

All server settings read from config.ini. CLI args override config.ini values.

[server]
host      = 0.0.0.0
port      = 8080
reload    = false
workers   = 4
log_level = info

[api]
cors_origins = https://myapp.com, https://admin.myapp.com

[logging]
level     = INFO
file_path = logs/service.log

[database]
url          = postgresql+asyncpg://user:pass@localhost/mydb
pool_size    = 10
max_overflow = 20

[cache]
enabled     = true
url         = redis://localhost:6379/0
default_ttl = 3600

Config priority order

VivekaAppConfig

Dataclass that defines the service identity and which infrastructure components to enable.

from viveka_server import VivekaAppConfig

config = VivekaAppConfig(
    name            = "order-service",
    version         = "2.1.0",
    description     = "Handles order lifecycle",
    enable_database = True,
    enable_cache    = True,
)

VivekaApp

Application lifecycle manager. Initializes logging, DB engine, and cache on construction. Override startup() and shutdown() for custom logic.

from viveka_server import VivekaApp, VivekaAppConfig

class MyApp(VivekaApp):

    async def startup(self) -> None:
        await super().startup()
        # load ML models, warm caches, seed data
        self.logger.info("Custom startup complete")

    async def shutdown(self) -> None:
        # flush buffers, close connections
        self.logger.info("Custom shutdown complete")
        await super().shutdown()

VivekaServer

Wraps FastAPI + Uvicorn. Registers middleware, exception handlers, and default routes. Pass a VivekaApp instance to the constructor.

Default Routes

Adding Custom Routes

from fastapi import APIRouter
from viveka_server import VivekaServer, VivekaApp, VivekaAppConfig

class OrderServer(VivekaServer):

    def build_routes(self, router: APIRouter) -> APIRouter:

        @router.get("/api/v1/orders")
        async def list_orders():
            return []

        @router.post("/api/v1/orders")
        async def create_order(order: OrderRequest):
            return await order_service.create(order)

        return router

app    = MyApp(VivekaAppConfig(name="order-service", version="1.0.0"))
server = OrderServer(app)
server.run()

Lifecycle Hooks

The server calls VivekaApp.startup() when FastAPI's lifespan starts and shutdown() when it ends. Always call super() to preserve base behaviour.

Server starts
    │
    ▼
FastAPI lifespan begins
    │── VivekaApp.startup()     ← your custom logic here
    ▼
Routes serve requests
    │
    ▼
Server shuts down (Ctrl+C / SIGTERM)
    └── VivekaApp.shutdown()    ← your custom cleanup here

Exception Handling

All VivekaAPIException subclasses are caught globally and returned as consistent JSON:

{
  "success": false,
  "error": {
    "code":    "NOT_FOUND",
    "message": "User 42 not found",
    "details": {}
  }
}

from viveka_server import NotFoundException, UnauthorizedException

@router.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_repo.get(user_id)
    if not user:
        raise NotFoundException(f"User {user_id} not found")
    return user

Database & Cache

app = VivekaApp(VivekaAppConfig(
    name            = "data-service",
    version         = "1.0.0",
    enable_database = True,   # reads [database] from config.ini
    enable_cache    = True,   # reads [cache] from config.ini
))

When enable_database=True, DatabaseSessionFactory (from viveka-kosha) is initialised and DbMiddleware is auto-registered on the FastAPI app — every HTTP request gets its own AsyncSession.

When enable_cache=True, VivekaCacheManager is initialised with RedisCacheService — the @cacheable and @cache_evict decorators from viveka-kosha work automatically.

CLI Arguments

python main.py --host 0.0.0.0 --port 8080 --workers 4
python main.py --reload
python main.py --log-level debug