dataknobs-common 1.3.9

Common utilities and base classes for dataknobs packages

dataknobs-common

Common utilities and base classes for all dataknobs packages.

Installation

pip install dataknobs-common

Overview

This package provides shared cross-cutting functionality used across all dataknobs packages:

• Exception Framework: Unified exception hierarchy with context support
• Registry Pattern: Generic registries for managing named items
• Serialization Protocol: Standard interfaces for to_dict/from_dict patterns

These patterns were extracted from common implementations across multiple packages to reduce duplication and provide consistency.

Features

1. Exception Framework

A unified exception hierarchy that all dataknobs packages extend. Supports both simple exceptions and context-rich exceptions with detailed error information.

Basic Usage

from dataknobs_common import DataknobsError, ValidationError, NotFoundError

# Simple exception
raise ValidationError("Invalid email format")

# Context-rich exception
raise NotFoundError(
    "User not found",
    context={"user_id": "123", "searched_in": "users_table"}
)

# Catch any dataknobs error
try:
    operation()
except DataknobsError as e:
    print(f"Error: {e}")
    if e.context:
        print(f"Context: {e.context}")

Available Exception Types

• DataknobsError - Base exception for all packages
• ValidationError - Data validation failures
• ConfigurationError - Configuration issues
• ResourceError - Resource acquisition/management failures
• NotFoundError - Item lookup failures
• OperationError - General operation failures
• ConcurrencyError - Concurrent operation conflicts
• SerializationError - Serialization/deserialization failures
• TimeoutError - Operation timeout errors

Package-Specific Extensions

from dataknobs_common import DataknobsError

class MyPackageError(DataknobsError):
    """Base exception for mypackage."""
    pass

class SpecificError(MyPackageError):
    """Specific error with custom context."""
    def __init__(self, item_id: str, message: str):
        super().__init__(
            f"Item '{item_id}': {message}",
            context={"item_id": item_id}
        )

2. Registry Pattern

Generic, thread-safe registries for managing collections of named items. Includes variants for caching and async support.

Basic Registry

from dataknobs_common import Registry

# Create a registry for tools
registry = Registry[Tool]("tools")

# Register items
registry.register("calculator", calculator_tool)
registry.register("search", search_tool, metadata={"version": "1.0"})

# Retrieve items
tool = registry.get("calculator")

# Check existence
if registry.has("search"):
    print("Search tool available")

# List all items
for key, tool in registry.items():
    print(f"{key}: {tool}")

# Get count
print(f"Registry has {registry.count()} tools")

Cached Registry

For items that should be cached with automatic TTL-based expiration:

from dataknobs_common import CachedRegistry

# Create registry with 5-minute cache
registry = CachedRegistry[Bot]("bots", cache_ttl=300)

# Get or create with factory
bot = registry.get_cached(
    "client1",
    factory=lambda: create_bot("client1")
)

# Get cache statistics
stats = registry.get_cache_stats()
print(f"Hit rate: {stats['hit_rate']:.2%}")

# Invalidate cache
registry.invalidate_cache("client1")  # Single item
registry.invalidate_cache()  # All items

Async Registry

For async contexts:

from dataknobs_common import AsyncRegistry

registry = AsyncRegistry[Resource]("resources")

# All operations are async
await registry.register("db", db_resource)
resource = await registry.get("db")
count = await registry.count()

Plugin Registry

For managing plugins with factory support, defaults, and lazy instantiation:

from dataknobs_common import PluginRegistry

# Define a base class
class Handler:
    def __init__(self, name: str, config: dict):
        self.name = name
        self.config = config

class DefaultHandler(Handler):
    pass

class CustomHandler(Handler):
    pass

# Create registry with default factory
registry = PluginRegistry[Handler]("handlers", default_factory=DefaultHandler)

# Register plugins
registry.register("custom", CustomHandler)

# Get instances (lazy creation with caching)
handler = registry.get("custom", config={"timeout": 30})
default = registry.get("unknown", config={})  # Uses default

# Async factory support
async def create_async_handler(name, config):
    handler = AsyncHandler(name, config)
    await handler.initialize()
    return handler

registry.register("async", create_async_handler)
handler = await registry.get_async("async", config={"url": "..."})

PluginRegistry Features

# Bulk registration
registry.bulk_register({
    "handler1": Handler1,
    "handler2": Handler2,
})

# Check registration
if registry.is_registered("custom"):
    print("Custom handler available")

# List all registered plugins
keys = registry.list_keys()

# Clear cached instances
registry.clear_cache("custom")  # Single
registry.clear_cache()  # All

# Get factory without creating instance
factory = registry.get_factory("custom")

# Set default after init
registry.set_default_factory(NewDefault)

Custom Registry Extensions

from dataknobs_common import Registry

class ToolRegistry(Registry[Tool]):
    """Registry for LLM tools."""

    def __init__(self):
        super().__init__("tools", enable_metrics=True)

    def register_tool(self, tool: Tool) -> None:
        """Register a tool with metadata."""
        self.register(
            tool.name,
            tool,
            metadata={"description": tool.description}
        )

    def get_by_category(self, category: str) -> list[Tool]:
        """Get all tools in a category."""
        return [
            tool for tool in self.list_items()
            if tool.category == category
        ]

3. Serialization Protocol

Standard protocol for objects that can be serialized to/from dictionaries.

Define Serializable Classes

from dataknobs_common import Serializable
from dataclasses import dataclass

@dataclass
class User:
    name: str
    email: str

    def to_dict(self) -> dict:
        return {"name": self.name, "email": self.email}

    @classmethod
    def from_dict(cls, data: dict) -> "User":
        return cls(name=data["name"], email=data["email"])

# Type checking works
user = User("Alice", "[e-mail]")
assert isinstance(user, Serializable)  # True

Serialization Utilities

from dataknobs_common import serialize, deserialize, serialize_list, deserialize_list

# Serialize single object
user = User("Alice", "[e-mail]")
data = serialize(user)
# {'name': 'Alice', 'email': '[e-mail]'}

# Deserialize
restored_user = deserialize(User, data)

# Serialize list
users = [User("Alice", "[e-mail]"), User("Bob", "[e-mail]")]
data_list = serialize_list(users)

# Deserialize list
restored_users = deserialize_list(User, data_list)

Type Checking

from dataknobs_common import is_serializable, is_deserializable

# Check if object can be serialized
if is_serializable(my_object):
    data = serialize(my_object)

# Check if class can deserialize
if is_deserializable(MyClass):
    obj = deserialize(MyClass, data)

Integration with Dataknobs Packages

This package is designed to be imported and extended by all dataknobs packages:

# In dataknobs_data package
from dataknobs_common import DataknobsError, NotFoundError

class DataknobsDataError(DataknobsError):
    """Base exception for data package."""
    pass

class RecordNotFoundError(NotFoundError):
    """Record not found in database."""
    pass

# In dataknobs_llm package
from dataknobs_common import Registry

class ToolRegistry(Registry[Tool]):
    """Registry for LLM tools."""

    def to_function_definitions(self) -> list[dict]:
        """Convert tools to function definitions."""
        return [tool.to_function_definition() for tool in self.list_items()]

Migration Guide

For Package Developers

If your package has custom exceptions, consider extending from dataknobs_common:

Before:

# packages/mypackage/exceptions.py
class MyPackageError(Exception):
    pass

After:

# packages/mypackage/exceptions.py
from dataknobs_common import DataknobsError

class MyPackageError(DataknobsError):
    pass

For Application Developers

You can now catch all dataknobs exceptions uniformly:

from dataknobs_common import DataknobsError

try:
    # Use any dataknobs package
    result = database.query()
    llm.complete()
    bot.chat()
except DataknobsError as e:
    logger.error(f"Dataknobs error: {e}")
    if e.context:
        logger.error(f"Context: {e.context}")

Why Common Package?

After building out core packages (data, llm, fsm, bots), we identified several patterns that were:

1. Implemented multiple times - Exception hierarchies in 4+ packages
2. Highly similar - Registry pattern ~100-150 lines duplicated 3 times
3. Genuinely cross-cutting - Used across domain boundaries

Extracting these patterns to dataknobs-common provides:

• Reduced duplication - Single source of truth
• Consistency - Same patterns everywhere
• Type safety - Shared protocols and interfaces
• Better error handling - Unified exception catching
• Ecosystem coherence - Packages feel integrated

API Reference

See module docstrings for detailed API documentation:

• dataknobs_common.exceptions - Exception hierarchy
• dataknobs_common.registry - Registry implementations
• dataknobs_common.serialization - Serialization protocols

Dependencies

• Python 3.10+
• No external dependencies (uses only standard library)

License

See LICENSE file in the root repository.