Core Shared Library

Overview

viveka-grantha solves three cross-cutting concerns so individual services don't have to rebuild them:

Installation

viveka-grantha has no internal Viveka dependencies. It can be installed standalone.

Runtime dependencies: redis>=5.0, hiredis>=2.0, aiohttp>=3.9

Full Startup Example

Wire all three modules together in a single bootstrap function called at application startup:

from viveka_common.config import ConfigService
from viveka_common.log import VivekaLogManager, VivekaLoggingConfig
from viveka_common.cache import VivekaCacheConfig, RedisCacheService, VivekaCacheManager

def bootstrap():
    # 1. Config
    config = ConfigService("config.ini")

    # 2. Logging
    VivekaLogManager.init(VivekaLoggingConfig(
        level     = config.get("logging.level", default="INFO"),
        file_path = config.get("logging.file_path", default="logs/app.log"),
    ))

    # 3. Cache
    cache_config = VivekaCacheConfig(
        enabled     = config.get("cache.enabled", default=False, data_type=bool),
        url         = config.get("cache.url", default="redis://localhost:6379/0"),
        default_ttl = config.get("cache.default_ttl", default=3600, data_type=int),
    )
    VivekaCacheManager.init(RedisCacheService(cache_config))

Configuration

config.ini Format

Configuration is read from a standard INI file with sections and keys:

[app]
name = my-service
env  = production

[logging]
level           = INFO
file_enabled    = true
file_path       = logs/app.log
console_enabled = true

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

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

ConfigService Usage

Access values using section.key dot notation. Priority order: environment variable → config.ini → default.

config = ConfigService()                        # reads config.ini from CWD
config = ConfigService("path/to/config.ini")    # custom path

# String (default)
name = config.get("app.name")

# Typed values
port    = config.get("api.port",    default=8000, data_type=int)
debug   = config.get("app.debug",   default=False, data_type=bool)
timeout = config.get("app.timeout", default=30.0, data_type=float)
hosts   = config.get("app.hosts",   data_type=list)  # comma-separated

# Read environment variable directly (database.url → DATABASE_URL)
token = config.get_env_variable("api.secret_key")

Supported Data Types

Logging

Initialization

Call VivekaLogManager.init() once at startup before any other code runs:

from viveka_common.log import VivekaLogManager, VivekaLoggingConfig

VivekaLogManager.init(VivekaLoggingConfig(
    level             = "INFO",
    console_enabled   = True,
    file_enabled      = True,
    file_path         = "logs/app.log",
    file_max_bytes    = 10_485_760,  # 10 MB
    file_backup_count = 5,
))

Getting a Logger

Any module or class gets its own named logger via get_instance(__name__):

from viveka_common.log import VivekaLogManager

_logger = VivekaLogManager.get_instance(__name__)

class UserService:
    async def create_user(self, name: str):
        _logger.info(f"Creating user: {name}")
        _logger.debug("Detailed debug info here")
        _logger.error("Something went wrong", exc_info=True)

VivekaLoggingConfig Options

Cache

Initialization

from viveka_common.cache import VivekaCacheConfig, RedisCacheService, VivekaCacheManager

config = VivekaCacheConfig(
    enabled     = True,
    url         = "redis://localhost:6379/0",
    default_ttl = 3600,
)
VivekaCacheManager.init(RedisCacheService(config))

To disable caching without changing any call sites, set enabled=False. All cache operations silently become no-ops.

VivekaCacheManager Methods

All methods are async. If the cache backend is unreachable, every method logs the error and returns a safe default — business logic is never interrupted.

from viveka_common.cache import VivekaCacheManager

# Store with optional TTL (seconds). Uses default_ttl if omitted.
await VivekaCacheManager.set("user:123", {"id": 123, "name": "Arjun"}, ttl=3600)

# Retrieve — returns None on cache miss
user = await VivekaCacheManager.get("user:123")

# Delete a single key
await VivekaCacheManager.delete("user:123")

# Delete all keys matching a glob pattern
await VivekaCacheManager.delete_pattern("user:*")

# Health check
alive = await VivekaCacheManager.ping()

Custom Cache Backend

Implement the VivekaCacheService abstract interface to plug in any backend:

from viveka_common.cache import VivekaCacheService, VivekaCacheManager

class InMemoryCacheService(VivekaCacheService):
    def __init__(self):
        self._store = {}

    async def get(self, key):    return self._store.get(key)
    async def set(self, key, value, ttl=None):
        self._store[key] = value; return True
    async def delete(self, key):
        self._store.pop(key, None); return True
    async def delete_pattern(self, pattern): return 0
    async def ping(self):         return True

VivekaCacheManager.init(InMemoryCacheService())