tessl/pypi-botbuilder-core

Microsoft Bot Framework Bot Builder core functionality for building conversational AI bots and chatbots in Python.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Storage

Name: tessl/pypi-botbuilder-core
Author: tessl

Storage implementations for persisting bot state and data. Includes memory storage for development and testing, plus abstract base classes for implementing custom storage solutions.

Capabilities

Storage Base Class

Abstract base class that defines the interface for all storage implementations in the Bot Framework, providing methods for reading, writing, and deleting data with support for concurrency control through eTags.

class Storage:
    async def read(self, keys):
        """
        Read items from storage.
        
        Args:
            keys (list): List of keys to read
            
        Returns:
            dict: Dictionary of key-value pairs for found items
        """
    
    async def write(self, changes):
        """
        Write items to storage.
        
        Args:
            changes (dict): Dictionary of key-value pairs to write
        """
    
    async def delete(self, keys):
        """
        Delete items from storage.
        
        Args:
            keys (list): List of keys to delete
        """

MemoryStorage

In-memory storage implementation for development and testing. Data is stored in memory and will be lost when the application restarts.

class MemoryStorage(Storage):
    def __init__(self, dictionary: dict = None):
        """
        Initialize memory storage.
        
        Args:
            dictionary (dict, optional): Optional initial dictionary
        """
    
    async def read(self, keys):
        """
        Read items from memory storage.
        
        Args:
            keys (list): List of keys to read
            
        Returns:
            dict: Dictionary of found items
        """
    
    async def write(self, changes):
        """
        Write items to memory storage.
        
        Args:
            changes (dict): Dictionary of changes to write
        """
    
    async def delete(self, keys):
        """
        Delete items from memory storage.
        
        Args:
            keys (list): List of keys to delete
        """

StoreItem

Base class for items that can be stored in bot storage, providing eTag support for concurrency control and change detection.

class StoreItem:
    def __init__(self):
        """Initialize store item with default eTag."""
        self.e_tag = "*"

QueueStorage

Queue-based storage implementation that provides FIFO (First-In-First-Out) storage capabilities for scenarios requiring ordered data processing.

class QueueStorage:
    def __init__(self, queues: dict = None):
        """
        Initialize queue storage.
        
        Args:
            queues (dict, optional): Optional initial queues dictionary
        """
    
    async def queue_activity(self, queue_name: str, activity):
        """
        Add activity to a queue.
        
        Args:
            queue_name (str): Name of the queue
            activity: Activity to queue
        """
    
    async def dequeue_activity(self, queue_name: str):
        """
        Remove and return activity from queue.
        
        Args:
            queue_name (str): Name of the queue
            
        Returns:
            Activity or None: Dequeued activity or None if queue is empty
        """
    
    async def get_queue_length(self, queue_name: str):
        """
        Get length of a queue.
        
        Args:
            queue_name (str): Name of the queue
            
        Returns:
            int: Number of items in queue
        """

Storage Utilities

Utility functions for working with storage implementations and data serialization.

def calculate_change_hash(obj):
    """
    Calculate hash for change detection.
    
    Args:
        obj: Object to calculate hash for
        
    Returns:
        str: Hash string for the object
    """

def sanitize_key(key: str):
    """
    Sanitize storage key to ensure compatibility.
    
    Args:
        key (str): Key to sanitize
        
    Returns:
        str: Sanitized key
    """

Usage Examples

Basic Memory Storage Setup

from botbuilder.core import MemoryStorage, ConversationState, UserState

# Create memory storage
memory_storage = MemoryStorage()

# Use with bot states
conversation_state = ConversationState(memory_storage)
user_state = UserState(memory_storage)

# Storage is now ready to use with your bot

Direct Storage Operations

from botbuilder.core import MemoryStorage

# Create storage
storage = MemoryStorage()

# Write data directly
changes = {
    "user:123": {"name": "John", "age": 30},
    "conversation:456": {"topic": "weather", "turn_count": 5}
}
await storage.write(changes)

# Read data back
keys = ["user:123", "conversation:456"]
data = await storage.read(keys)

print(f"User data: {data.get('user:123')}")
print(f"Conversation data: {data.get('conversation:456')}")

# Delete data
await storage.delete(["user:123"])

Custom Storage Implementation

from botbuilder.core import Storage
import json
import os

class FileStorage(Storage):
    def __init__(self, directory: str):
        self.directory = directory
        os.makedirs(directory, exist_ok=True)
    
    async def read(self, keys):
        items = {}
        for key in keys:
            file_path = os.path.join(self.directory, f"{key}.json")
            if os.path.exists(file_path):
                with open(file_path, 'r') as f:
                    items[key] = json.load(f)
        return items
    
    async def write(self, changes):
        for key, value in changes.items():
            file_path = os.path.join(self.directory, f"{key}.json")
            with open(file_path, 'w') as f:
                json.dump(value, f, indent=2)
    
    async def delete(self, keys):
        for key in keys:
            file_path = os.path.join(self.directory, f"{key}.json")
            if os.path.exists(file_path):
                os.remove(file_path)

# Usage
file_storage = FileStorage("./bot_data")
conversation_state = ConversationState(file_storage)
user_state = UserState(file_storage)

Storage with eTag Support

from botbuilder.core import StoreItem, MemoryStorage

class UserProfile(StoreItem):
    def __init__(self):
        super().__init__()
        self.name = None
        self.email = None
        self.preferences = {}

# Usage with eTag handling
storage = MemoryStorage()

# Create and store user profile
user_profile = UserProfile()
user_profile.name = "Alice"
user_profile.email = "alice@example.com"

await storage.write({"user:alice": user_profile})

# Read back with eTag preserved
data = await storage.read(["user:alice"])
retrieved_profile = data["user:alice"]

print(f"Profile eTag: {retrieved_profile.e_tag}")

# Modify and save (eTag will be updated automatically)
retrieved_profile.preferences["theme"] = "dark"
await storage.write({"user:alice": retrieved_profile})

Queue Storage Usage

from botbuilder.core import QueueStorage, Activity

# Create queue storage
queue_storage = QueueStorage()

# Queue some activities
activity1 = Activity(type="message", text="First message")
activity2 = Activity(type="message", text="Second message")

await queue_storage.queue_activity("pending", activity1)
await queue_storage.queue_activity("pending", activity2)

# Check queue length
length = await queue_storage.get_queue_length("pending")
print(f"Queue length: {length}")

# Process queue
while await queue_storage.get_queue_length("pending") > 0:
    activity = await queue_storage.dequeue_activity("pending")
    print(f"Processing: {activity.text}")

Storage Error Handling

async def safe_storage_operation(storage, key, data):
    try:
        await storage.write({key: data})
        print(f"Successfully stored data for key: {key}")
    except Exception as e:
        print(f"Storage error for key {key}: {e}")
        # Implement retry logic or fallback storage
        await fallback_storage_operation(key, data)

async def fallback_storage_operation(key, data):
    # Fallback to a different storage mechanism
    fallback_storage = MemoryStorage()
    await fallback_storage.write({key: data})
    print(f"Data stored in fallback storage for key: {key}")

Batch Storage Operations

async def batch_user_update(storage, user_updates):
    """Update multiple users in a single storage operation."""
    
    # Prepare batch changes
    changes = {}
    for user_id, update_data in user_updates.items():
        key = f"user:{user_id}"
        changes[key] = update_data
    
    # Write all changes at once
    await storage.write(changes)
    print(f"Updated {len(changes)} users in batch")

# Usage
user_updates = {
    "123": {"name": "John", "last_seen": "2023-01-01"},
    "456": {"name": "Jane", "last_seen": "2023-01-02"},
    "789": {"name": "Bob", "last_seen": "2023-01-03"}
}

await batch_user_update(storage, user_updates)

Storage Key Management

from botbuilder.core import sanitize_key

def create_storage_key(prefix: str, identifier: str, suffix: str = None):
    """Create a properly formatted storage key."""
    parts = [prefix, identifier]
    if suffix:
        parts.append(suffix)
    
    key = ":".join(parts)
    return sanitize_key(key)

# Usage
user_key = create_storage_key("user", "john.doe@example.com")
conversation_key = create_storage_key("conversation", "teams-channel-123", "thread-456")

print(f"User key: {user_key}")
print(f"Conversation key: {conversation_key}")

Types

class StorageKeyFactory:
    """Factory for creating consistent storage keys."""
    
    @staticmethod
    def user_key(user_id: str):
        """Create user storage key."""
        return f"user:{user_id}"
    
    @staticmethod
    def conversation_key(conversation_id: str):
        """Create conversation storage key."""
        return f"conversation:{conversation_id}"
    
    @staticmethod
    def private_conversation_key(user_id: str, conversation_id: str):
        """Create private conversation storage key."""
        return f"private:{user_id}:{conversation_id}"

Install with Tessl CLI