vs-common

Core shared library for Viveka Sutra — config, logging, and async cache management.

---

Installation

pip install vs-common

With Redis support:

pip install vs-common[redis]

---

Application Startup

vs-common components are designed to be initialised once at application startup, in this order:

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 application 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, but in production you should always call bootstrap() explicitly so configuration is intentional and traceable.

Re-initialisation behaviour:

Component	What happens
VsLogManager.init()	Updates log level on all existing handlers. Handlers and formatters from first init are kept. Safe to call again.
VsCacheManager.init()	Replaces the active backend with a new instance. All data in the previous cache is lost — including all in-memory entries if using InMemoryCache.
VsIniConfig	Not a singleton. Each VsIniConfig(path) creates a fresh instance and re-reads the file. Use config.reload() to refresh an existing instance in-place instead.

---

Modules

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("error message", exc_info=True)
logger.exception("exception with traceback")

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.

In-memory (default):

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:

from vs_common.cache.vs_cache_manager import VsCacheManager
from vs_common.schema.vs_cache_config import VsCacheConfig

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

---

Customization

Extending the Config

Implement VsBaseConfig to load configuration from any source — YAML, AWS SSM, environment-only, 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 it directly wherever a VsBaseConfig is expected:

config = YamlConfig("config.yaml")
config.get("database.url")
config.get("database.port", data_type=int)

get_section() is optional — if your backend does not support sections, the base class returns {} by default.

---

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:

from vs_common.cache.vs_cache_manager import VsCacheManager

VsCacheManager.register(MyCustomCache())

All subsequent calls to VsCacheManager.get/set/delete will use your implementation.

RedisCache reference implementation

RedisCache ships with the library and serves as a reference for how to implement VsBaseCache against an async client:

from typing import Any, Optional
import redis.asyncio as aioredis
from vs_common.cache.vs_base_cache import VsBaseCache
from vs_common.schema.vs_cache_config import VsCacheConfig

class RedisCache(VsBaseCache):

    def __init__(self, config: VsCacheConfig):
        self._default_ttl = config.default_ttl
        self._client = aioredis.Redis(
            host=config.host,
            port=config.port,
            db=config.db,
            password=config.password,
            decode_responses=False,
        )

    async def get(self, key: str) -> Optional[Any]:
        import pickle
        value = await self._client.get(key)
        return pickle.loads(value) if value is not None else None

    async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> bool:
        import pickle
        ex = ttl if ttl is not None else self._default_ttl
        return await self._client.set(key, pickle.dumps(value), ex=ex)

    async def delete(self, key: str) -> bool:
        return await self._client.delete(key) > 0

    async def delete_pattern(self, pattern: str) -> int:
        keys = await self._client.keys(pattern)
        if not keys:
            return 0
        return await self._client.delete(*keys)

    async def ping(self) -> bool:
        try:
            return await self._client.ping()
        except Exception:
            return False

    async def close(self) -> None:
        await self._client.aclose()

---

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 as e:
            self.vs_logger.error(f"Order {order_id} failed", exc_info=True)

The string passed to @vs_logger becomes the logger name — it appears in every log line and lets you filter logs by service/component.

Use cases:

Scenario	Example name
Service class	@vs_logger("PaymentService")
Repository / DAO	@vs_logger("UserRepository")
Background worker	@vs_logger("IngestWorker")
Cache manager	@vs_logger("VsCacheManager")

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")

exception() automatically captures and appends the current exception traceback, making it ideal inside except blocks:

try:
    result = call_external_api()
except Exception:
    self.vs_logger.exception("External API call failed")

Passing structured context:

All methods accept an optional extra dict that gets merged into the log record. Useful for attaching request IDs, user IDs, or trace 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 immediately — the method body never runs.

from vs_common.decorator.vs_cache_decorator import cacheable

class ProductService:

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

How the cache key is built:

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

Why it makes life easy:

Without @cacheable you write the same boilerplate on every method — check cache, return if hit, call db, store result, return. With it, the entire pattern collapses to one line above the method signature. TTL is declared where the method is, so the caching intent is always visible alongside the logic it protects.

# without @cacheable — repeated everywhere
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 — all of the above, in one line
@cacheable(key_prefix="product", ttl=300)
async def get_product(self, product_id: str):
    return await db.fetch_product(product_id)

---

@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)

When update_product or delete_product is called, all product:* keys are wiped — so the next get_product call fetches fresh data from the database.

Why it makes life easy:

Cache invalidation is one of the hardest problems in software. @cache_evict makes it declarative — you say what to invalidate, not how. No risk of forgetting to clear cache after a write, no scattered VsCacheManager.delete_pattern() calls across your codebase.

Typical pairing pattern:

@cacheable(key_prefix="user", ttl=600)
async def get_user(self, user_id: str): ...

@cache_evict(key_prefix="user")
async def update_user(self, user_id: str, data: dict): ...

@cache_evict(key_prefix="user")
async def delete_user(self, user_id: str): ...

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

---

Running Tests

./run_tests.sh