vs-common

Core shared library for Viveka Sutra — config, logging, cache management, decorators, exceptions, and trace publishing.

---

Overview

vs-common is the foundation every VS service and agent is built on. It provides config loading, structured logging, async cache management, utility decorators, a base exception hierarchy, and a trace publishing interface — all initialised once at startup and shared across the application.

---

The Problem It Solves

Every service needs config, logging, and cache. Without a shared library, each service reinvents the same boilerplate.

Without vs-common

import logging
import os

# every service does this differently
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("MyService")

def get_db_url():
    return os.getenv("DATABASE_URL", "postgres://localhost/mydb")

# cache? write it yourself, or skip it

With vs-common

config = VsIniConfig("config.ini")
VsLogManager.init(VsLogConfig(level=config.get("logging.level")))
VsCacheManager.init(VsCacheConfig(backend=config.get("cache.backend")))

One pattern, every service, zero drift.

---

Installation

pip install vs-common

With Redis support:

pip install vs-common[redis]

---

Dependencies

Library	Required	Purpose
Python stdlib	Yes	configparser, logging, asyncio
redis	No — install with [redis] extra	Redis cache backend via RedisCache

---

Configuration

All keys are optional — each component falls back to safe defaults if not configured.

Key	Default	Description
logging.level	INFO	Log level: DEBUG, INFO, WARN, ERROR
logging.file_path	logs/viveka.log	Log file path
logging.console_enabled	True	Enable console output
logging.file_enabled	True	Enable file output
logging.file_max_bytes	10485760	Max log file size before rotation (10 MB)
logging.file_backup_count	5	Number of rotated files to keep
cache.enabled	False	Enable caching
cache.backend	memory	Backend: memory or redis
cache.host	localhost	Redis host
cache.port	6379	Redis port
cache.db	0	Redis DB index
cache.password	None	Redis password
cache.default_ttl	3600	Default TTL in seconds

---

Quick Start

Initialise all components once at application startup:

from vs_common.config.vs_ini_config import VsIniConfig
from vs_common.log.vs_log_manager import VsLogManager
from vs_common.cache.vs_cache_manager import VsCacheManager
from vs_common.schema.vs_log_config import VsLogConfig
from vs_common.schema.vs_cache_config import VsCacheConfig

def bootstrap():
    config = VsIniConfig("config.ini")

    VsLogManager.init(VsLogConfig(
        level=config.get("logging.level", default="INFO"),
        file_path=config.get("logging.file_path", default="logs/app.log"),
    ))

    VsCacheManager.init(VsCacheConfig(
        enabled=config.get("cache.enabled", data_type=bool, default=False),
        backend=config.get("cache.backend", default="memory"),
        host=config.get("cache.host", default="localhost"),
        port=config.get("cache.port", data_type=int, default=6379),
        password=config.get("cache.password"),
    ))

Call bootstrap() once at your entry point before anything else runs.

What happens if you skip initialisation:

Component	Behaviour without init()
VsLogManager	Auto-inits with INFO level, console + file enabled, default log path
VsCacheManager	Auto-inits with InMemoryCache on first use
VsIniConfig	Reads config.ini from current directory; missing file falls back to env vars and defaults

Skipping is safe for quick scripts and tests. In production, always call bootstrap() explicitly.

Re-initialisation behaviour:

Component	What happens
VsLogManager.init()	Updates log level on all existing handlers. Safe to call again.
VsCacheManager.init()	Replaces the active backend. All data in the previous cache is lost.
VsIniConfig	Not a singleton. Each VsIniConfig(path) creates a fresh instance. Use config.reload() to refresh in-place.

---

Config

VsIniConfig reads from a config.ini file and supports environment variable overrides.

from vs_common.config.vs_ini_config import VsIniConfig

config = VsIniConfig("config.ini")

config.get("database.url")                        # str
config.get("database.port", data_type=int)        # int
config.get("app.debug", data_type=bool)           # bool
config.get("app.tags", data_type=list)            # list
config.get("app.missing", default="fallback")     # default if not found
config.get_section("database")                    # dict of all keys in section
config.reload()                                   # re-read config.ini from disk

Environment variables take priority over config.ini. Key "database.url" maps to env var DATABASE_URL.

config.ini example:

[database]
url = postgres://localhost/mydb
port = 5432

[app]
debug = true
tags = a,b,c

---

Logging

VsLogManager is a singleton that configures the root logger once at startup.

from vs_common.log.vs_log_manager import VsLogManager
from vs_common.schema.vs_log_config import VsLogConfig

VsLogManager.init(VsLogConfig(
    level="DEBUG",
    console_enabled=True,
    file_enabled=True,
    file_path="logs/app.log",
))

logger = VsLogManager.get_instance("my-service")

logger.debug("debug message")
logger.info("info message")
logger.warn("warning message")
logger.error("failure that needs attention")
logger.exception("error with full traceback — use inside except blocks")

VsLogConfig defaults:

Field	Default
level	INFO
console_enabled	True
file_enabled	True
file_path	logs/viveka.log
file_max_bytes	10 MB
file_backup_count	5

If VsLogManager.get_instance() is called before init(), it auto-initialises with defaults.

---

Cache

VsCacheManager is a static facade over a pluggable cache backend.

from vs_common.cache.vs_cache_manager import VsCacheManager

await VsCacheManager.set("key", {"data": 1}, ttl=60)
value = await VsCacheManager.get("key")
await VsCacheManager.delete("key")
await VsCacheManager.delete_pattern("user:*")
alive = await VsCacheManager.ping()

Redis backend:

VsCacheManager.init(VsCacheConfig(
    enabled=True,
    backend="redis",
    host="localhost",
    port=6379,
    db=0,
    password="secret",
    default_ttl=3600,
))

VsCacheConfig defaults:

Field	Default
enabled	False
backend	memory
host	localhost
port	6379
db	0
password	None
default_ttl	3600

---

Decorators

@vs_logger

Injects a VsLogger instance as a class attribute, giving any class structured logging without manual setup.

from vs_common.decorator.vs_logger_decorator import vs_logger

@vs_logger("OrderService")
class OrderService:

    def place_order(self, order_id: str):
        self.vs_logger.info(f"Placing order {order_id}")
        try:
            ...
        except Exception:
            self.vs_logger.exception(f"Order {order_id} failed")

The string passed to @vs_logger becomes the logger name and appears in every log line.

Available log methods:

self.vs_logger.debug("low-level detail")
self.vs_logger.info("normal operations")
self.vs_logger.warn("something unexpected but recoverable")
self.vs_logger.error("failure that needs attention")
self.vs_logger.exception("error with full traceback — use inside except blocks")

Passing structured context:

self.vs_logger.info("User logged in", extra={"user_id": user.id, "ip": request.ip})

---

@cacheable

Caches the return value of an async method. On the first call the method executes and the result is stored. On subsequent calls with the same arguments the cached value is returned — the method body never runs.

Without @cacheable

async def get_product(self, product_id: str):
    cached = await VsCacheManager.get(f"product:{product_id}")
    if cached:
        return cached
    result = await db.fetch_product(product_id)
    await VsCacheManager.set(f"product:{product_id}", result, ttl=300)
    return result

With @cacheable

from vs_common.decorator.vs_cache_decorator import cacheable

@cacheable(key_prefix="product", ttl=300)
async def get_product(self, product_id: str):
    return await db.fetch_product(product_id)

Parameters:

Parameter	Required	Description
key_prefix	Yes	Prefix for the cache key
ttl	No	TTL in seconds; falls back to VsCacheConfig.default_ttl

Cache key format:

Call	Cache key
get_product("abc")	product:abc
get_product("abc", locale="en")	product:abc:locale=en
get_product()	product:default

---

@cache_evict

After a write/update/delete method runs, deletes all cache keys matching key_prefix:*. Keeps the cache consistent without manual invalidation calls.

from vs_common.decorator.vs_cache_decorator import cacheable, cache_evict

class ProductService:

    @cacheable(key_prefix="product", ttl=300)
    async def get_product(self, product_id: str):
        return await db.fetch_product(product_id)

    @cache_evict(key_prefix="product")
    async def update_product(self, product_id: str, data: dict):
        return await db.update_product(product_id, data)

    @cache_evict(key_prefix="product")
    async def delete_product(self, product_id: str):
        return await db.delete_product(product_id)

Every read is cached. Every write auto-invalidates. Zero manual cache management.

Parameters:

Parameter	Required	Description
key_prefix	Yes	Deletes all keys matching key_prefix:* after the method runs

---

Extending vs-common

Extending the Config

Implement VsBaseConfig to load configuration from any source — YAML, AWS SSM, Vault, etc.

from typing import Any, Dict, Type
from vs_common.config.vs_base_config import VsBaseConfig

class YamlConfig(VsBaseConfig):

    def __init__(self, path: str):
        import yaml
        with open(path) as f:
            self._data = yaml.safe_load(f)

    def get(self, key: str, default: Any = None, data_type: Type = str) -> Any:
        section, config_key = key.split(".", 1)
        value = self._data.get(section, {}).get(config_key)
        if value is None:
            return default
        return data_type(value)

    def get_section(self, section: str) -> Dict[str, str]:
        return self._data.get(section, {})

Use VsBaseConfig as the type hint everywhere so implementations are interchangeable.

Notes:

• get_section() is optional — the base class returns {} by default if not implemented.
• VsIniConfig is the built-in implementation and is sufficient for most services.

---

Extending the Cache

Implement VsBaseCache to plug in any cache backend — DynamoDB, Memcached, a custom store, etc.

from typing import Any, Optional
from vs_common.cache.vs_base_cache import VsBaseCache

class MyCustomCache(VsBaseCache):

    async def get(self, key: str) -> Optional[Any]: ...
    async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> bool: ...
    async def delete(self, key: str) -> bool: ...
    async def delete_pattern(self, pattern: str) -> int: ...
    async def ping(self) -> bool: ...

Register it with the manager:

VsCacheManager.register(MyCustomCache())

RedisCache (ships with the library) is the reference implementation — read it to understand how to implement VsBaseCache against an async client.

---

Extending the Trace Publisher

Implement VsBaseTracePublisher to stream trace events from graph nodes and agent pipelines to any target — WebSocket, message queue, logging, etc.

from vs_common.publisher.vs_base_trace_publisher import VsBaseTracePublisher
from vs_common.schema.vs_trace_level import VsTraceLevel

class MyTracePublisher(VsBaseTracePublisher):

    async def publish(
        self,
        trace_id: str,
        message: str,
        level: VsTraceLevel = VsTraceLevel.INFO,
        data: dict = None,
    ) -> None:
        print(f"[{level}] {trace_id}: {message}", data)

Pass an instance to VsBaseGraph or RagPipeline at construction time:

graph = OrchestratorGraph(trace_publisher=MyTracePublisher())

VsTraceLevel values:

Level	Value
VsTraceLevel.DEBUG	"debug"
VsTraceLevel.INFO	"info"
VsTraceLevel.WARN	"warn"
VsTraceLevel.ERROR	"error"

---

Error Handling

All VS libraries raise exceptions that extend VsBaseException. Catch by base type to handle errors uniformly across services.

from vs_common.exception.vs_base_exception import VsBaseException

try:
    ...
except VsBaseException as e:
    logger.error(e.message)

Define custom exceptions by extending VsBaseException:

class MyServiceException(VsBaseException):
    pass

raise MyServiceException("Something went wrong")

All domain-specific exceptions in VS libraries (e.g. VsAuthenticationError in vs-security) extend VsBaseException.

---

Class Reference

---

VsIniConfig

Reads configuration from a .ini file. Supports environment variable overrides and typed value casting.

Constructor:

Parameter	Type	Required	Description
path	str	Yes	Path to the .ini file

Methods:

Method	Signature	Description
get	get(key: str, default: Any = None, data_type: Type = str) -> Any	Get a value by dotted key (e.g. "database.url"). Casts to data_type. Returns default if not found.
get_section	get_section(section: str) -> Dict[str, str]	Returns all key-value pairs under a section as a dict.
reload	reload() -> None	Re-reads the config file from disk in-place.

Notes:

• Environment variables take priority over config.ini. Key "database.url" maps to env var DATABASE_URL.
• data_type supports: str, int, float, bool, list.
• list splits on commas: "a,b,c" → ["a", "b", "c"].

---

VsBaseConfig

Abstract base class for all configuration sources. Implement this to load config from any backend (YAML, AWS SSM, Vault, etc.).

Methods:

Method	Signature	Description
get	get(key: str, default: Any = None, data_type: Type = str) -> Any	Abstract. Get a value by dotted key.
get_section	get_section(section: str) -> Dict[str, str]	Optional. Base returns {}.

Notes:

• Use VsBaseConfig as the type hint everywhere so implementations are interchangeable.
• VsIniConfig is the built-in implementation.

---

VsLogManager

Singleton that configures the root logger once at startup. All loggers created after init() inherit the configured level and handlers.

Methods:

Method	Signature	Description
init	init(config: VsLogConfig) -> None	Class method. Initialise the root logger with the given config. Safe to call again — updates level on existing handlers.
get_instance	get_instance(name: str) -> VsLogger	Class method. Returns a named logger instance. Auto-inits with defaults if init() has not been called.

---

VsLogConfig

Configuration schema for VsLogManager.

Fields:

Field	Type	Default	Description
level	str	INFO	Log level: DEBUG, INFO, WARN, ERROR
console_enabled	bool	True	Enable console output
file_enabled	bool	True	Enable file output
file_path	str	logs/viveka.log	Log file path
file_max_bytes	int	10485760	Max file size before rotation (10 MB)
file_backup_count	int	5	Number of rotated files to keep

---

VsCacheManager

Static facade over a pluggable cache backend. Use when you need to read/write cache without coupling to a specific backend.

Methods:

Method	Signature	Description
init	init(config: VsCacheConfig) -> None	Class method. Initialise with the given backend config. Replaces any existing backend — all cached data is lost.
register	register(backend: VsBaseCache) -> None	Class method. Register a custom cache backend directly.
get	get(key: str) -> Optional[Any]	Async. Fetch a value by key. Returns None if not found.
set	set(key: str, value: Any, ttl: Optional[int] = None) -> bool	Async. Store a value. Uses VsCacheConfig.default_ttl if ttl is not provided.
delete	delete(key: str) -> bool	Async. Remove a single key. Returns True if deleted.
delete_pattern	delete_pattern(pattern: str) -> int	Async. Remove all keys matching pattern (e.g. user:*). Returns count of deleted keys.
ping	ping() -> bool	Async. Returns True if the backend is reachable.

---

VsCacheConfig

Configuration schema for VsCacheManager.

Fields:

Field	Type	Default	Description
enabled	bool	False	Enable caching. If False, all operations are no-ops.
backend	str	memory	Backend type: memory or redis
host	str	localhost	Redis host
port	int	6379	Redis port
db	int	0	Redis DB index
password	str	None	Redis password
default_ttl	int	3600	Default TTL in seconds when ttl is not specified

---

VsBaseCache

Abstract base class for cache backends. Implement this to plug in any cache — DynamoDB, Memcached, a custom store, etc.

Methods:

Method	Signature	Description
get	get(key: str) -> Optional[Any]	Async abstract. Fetch a value by key.
set	set(key: str, value: Any, ttl: Optional[int] = None) -> bool	Async abstract. Store a value with optional TTL.
delete	delete(key: str) -> bool	Async abstract. Remove a key.
delete_pattern	delete_pattern(pattern: str) -> int	Async abstract. Remove all keys matching pattern.
ping	ping() -> bool	Async abstract. Health check.

Notes:

• Register with VsCacheManager.register(MyCustomCache()).
• RedisCache (ships with the library) is the reference implementation.

---

VsBaseException

Base exception class for all Viveka Sutra custom exceptions.

Constructor:

Parameter	Type	Required	Description
message	str	Yes	Human-readable error message

Attributes:

Attribute	Type	Description
message	str	The error message passed at construction

Notes:

• All domain-specific exceptions in VS libraries extend VsBaseException (e.g. VsAuthenticationError in vs-security).
• Catch by base type to handle errors uniformly: except VsBaseException as e.

---

VsBaseTracePublisher

Abstract base class for publishing trace events from graph nodes and agent pipelines.

Methods:

Method	Signature	Description
publish	publish(trace_id: str, message: str, level: VsTraceLevel = VsTraceLevel.INFO, data: dict = None) -> None	Async abstract. Publish a trace event.

Notes:

• Pass an instance to VsBaseGraph or RagPipeline at construction time.
• Used by vs-graph to stream execution traces in real time.

---

VsTraceLevel

Enum of trace severity levels used by VsBaseTracePublisher.

Values:

Value	String
VsTraceLevel.DEBUG	"debug"
VsTraceLevel.INFO	"info"
VsTraceLevel.WARN	"warn"
VsTraceLevel.ERROR	"error"

---

@vs_logger

Decorator. Injects a VsLogger instance as self.vs_logger on any class.

Parameters:

Parameter	Type	Required	Description
name	str	Yes	Logger name — appears in every log line

Methods on VsLogger:

Method	Signature	Description
debug	debug(msg: str, extra: dict = None) -> None	Low-level detail
info	info(msg: str, extra: dict = None) -> None	Normal operations
warn	warn(msg: str, extra: dict = None) -> None	Unexpected but recoverable
error	error(msg: str, extra: dict = None) -> None	Failure that needs attention
exception	exception(msg: str, extra: dict = None) -> None	Error with full traceback — use inside except blocks

---

@cacheable

Decorator. Caches the return value of an async method by key. On cache hit the method body is skipped entirely.

Parameters:

Parameter	Type	Required	Description
key_prefix	str	Yes	Prefix for the cache key
ttl	int	No	TTL in seconds. Falls back to VsCacheConfig.default_ttl.

---

@cache_evict

Decorator. After the decorated async method runs, deletes all cache keys matching key_prefix:*.

Parameters:

Parameter	Type	Required	Description
key_prefix	str	Yes	Deletes all keys matching key_prefix:* after the method completes

---

Running Tests

./run_tests.sh