VivekaSutra delivers production-ready Python libraries and agentic AI frameworks so your team ships faster — not fights infrastructure.
What We Do
We design and build the foundational layers that AI-powered services depend on — so developers can focus on intelligence, not plumbing.
Composable, observable agent runtimes built for production. Design multi-step reasoning loops, tool-use pipelines, and autonomous workflows with confidence.
Every library we ship is built async from the ground up — FastAPI, SQLAlchemy async, Redis asyncio. No blocking, no surprises under load.
Small, focused packages with clear responsibilities. Add what you need, leave what you don't. Each library works standalone or as part of the Viveka ecosystem.
Connection pooling, rotating log files, cache fallbacks, transaction rollbacks — the operational concerns are handled so you never have to think about them.
Familiar decorator-based APIs inspired by Spring Boot — @entity, @transactional, @cacheable — so teams coming from Java feel right at home.
Published on PyPI. Install in one line, integrate in minutes. Full type hints, py.typed markers, and IDE autocomplete out of the box.
Libraries
Three focused libraries that form the backbone of any Viveka-powered service. Free to download and use — proprietary, no warranty.
The core shared library. Provides configuration management, structured logging, and async Redis caching — the three cross-cutting concerns every service needs on day one.
A Spring JPA-inspired async database library. Declarative entity mapping, generic repositories, transaction management, and cache decorators — built on SQLAlchemy async and Redis.
FastAPI + Uvicorn application factory. Spin up a production-ready API server with built-in health checks, CORS, exception handling, DB middleware, and lifecycle hooks — in under ten lines of code.
Live Demos
Three self-contained FastAPI services showing how to use viveka-db-lib and viveka-server-lib in real applications. Clone the repo and run any demo in one command.
The simplest demo. One entity, one repository, full CRUD. The recommended starting point to understand how viveka-db-lib connects models, repositories, and the automatic session lifecycle.
@entity + @table — declarative entity mappingVivekaDbBase — shared SQLAlchemy declarative baseDbRepo[T] — insert, get, update, delete, count built-in@repository — marks the repository class@transactional — session injected automatically per callVivekaApp.startup() — creates DB tables on bootVivekaServer.build_routes() — wires FastAPI routes@entity
@table("todos")
class Todo(VivekaDbBase):
id = mapped_column(Integer, primary_key=True)
title = mapped_column(String(200), nullable=False)
completed = mapped_column(Boolean, default=False)
created_at = mapped_column(DateTime, default=datetime.utcnow)
@repository
class TodoRepo(DbRepo[Todo]):
@transactional
async def save(self, todo: Todo) -> Todo:
return (await self.insert_all([todo]))[0]
@transactional
async def complete(self, todo_id: int):
todo = await self.get_by_id(todo_id)
if todo:
todo.completed = True
return await self.update(todo)
Two related entities — Department and Employee. Shows how viveka-db-lib handles foreign key relationships, field-based queries, and paginated listing across two repos sharing a single service.
ForeignKey relationship — Employee.department_id → departments.idfind_all_by_field — fetch all employees in a departmentfind_by_field — lookup employee by emailget_all(skip, limit) — built-in paginationcount() — total record count without loading rowsVivekaServer@entity
@table("employees")
class Employee(VivekaDbBase):
id = mapped_column(Integer, primary_key=True)
name = mapped_column(String(200), nullable=False)
role = mapped_column(String(100), nullable=False)
department_id = mapped_column(Integer, ForeignKey("departments.id"))
@repository
class EmployeeRepo(DbRepo[Employee]):
@transactional
async def find_by_department(self, dept_id: int):
return await self.find_all_by_field("department_id", dept_id)
@transactional
async def list_all(self, skip=0, limit=100):
return await self.get_all(skip=skip, limit=limit)
An e-commerce catalog with categories and products. The most feature-rich demo — focuses on the @cacheable and @cache_evict decorators. Works without Redis via graceful degradation; enable caching by uncommenting the Redis config.
@cacheable(key_prefix, ttl) — cache-aside read on GET /products/{id}@cache_evict(key_prefix) — invalidates cache on price/stock updatefind_all_by_field — products by categoryForeignKey relationship — Product → Category@entity
@table("products")
class Product(VivekaDbBase):
id = mapped_column(Integer, primary_key=True)
name = mapped_column(String(200), nullable=False)
price = mapped_column(Float, nullable=False)
category_id = mapped_column(Integer, ForeignKey("categories.id"))
@repository
class ProductRepo(DbRepo[Product]):
@transactional
@cacheable(key_prefix="product", ttl=1800)
async def get(self, product_id: int):
return await self.get_by_id(product_id)
@transactional
@cache_evict(key_prefix="product")
async def update_product(self, product: Product):
return await self.update(product)
Technology
Every technology choice is deliberate — async-native, production-proven, and widely adopted in the Python ecosystem.
Get Started
All three libraries are on PyPI. Install them in seconds and have your service infrastructure ready in minutes.