HCFS/examples/basic_usage.py

#!/usr/bin/env python3
"""
HCFS Basic Usage Example

This example demonstrates the core HCFS API for context navigation,
storage, and retrieval.
"""

import asyncio
from datetime import datetime
from hcfs.api import ContextAPI, ContextBlob


async def basic_example():
    """Basic HCFS usage example."""
    
    # Initialize HCFS API
    api = ContextAPI()
    
    print("=== HCFS Basic Usage Example ===\n")
    
    # 1. Navigation
    print("1. Navigation")
    print(f"Current path: {api.context_pwd()}")
    
    # Create a project structure
    await api.context_cd("/")
    print(f"Changed to root: {api.context_pwd()}")
    
    await api.context_cd("/project")
    print(f"Changed to project: {api.context_pwd()}")
    
    # 2. Creating context
    print("\n2. Creating Context")
    
    # Add project-level context
    project_context = ContextBlob(
        content="AI-powered web application for task automation",
        content_type="documentation",
        tags=["project", "ai", "automation"],
        metadata={
            "priority": "high",
            "team": "ai-research",
            "status": "active"
        }
    )
    
    blob_id = await api.context_push("/project", project_context)
    print(f"Created project context: {blob_id}")
    
    # Add source code context
    src_context = ContextBlob(
        content="Main application source code. Uses FastAPI for REST API, SQLAlchemy for ORM, and Pydantic for data validation.",
        content_type="code",
        tags=["fastapi", "sqlalchemy", "pydantic", "backend"],
        metadata={
            "language": "python",
            "framework": "fastapi"
        }
    )
    
    await api.context_cd("/project/src")
    blob_id = await api.context_push("/project/src", src_context)
    print(f"Created src context: {blob_id}")
    
    # Add specific module context
    auth_context = ContextBlob(
        content="Authentication module using JWT tokens. Implements login, logout, and token refresh endpoints.",
        content_type="code", 
        tags=["auth", "jwt", "security", "endpoints"],
        metadata={
            "module": "auth.py",
            "security_level": "high"
        }
    )
    
    blob_id = await api.context_push("/project/src/auth.py", auth_context)
    print(f"Created auth module context: {blob_id}")
    
    # 3. Context retrieval
    print("\n3. Context Retrieval")
    
    # Get context from current path
    await api.context_cd("/project/src/auth.py")
    current_context = await api.context_get(depth=1)
    print(f"Current path context ({len(current_context)} blobs):")
    for blob in current_context:
        print(f"  - {blob.content_type}: {blob.content[:50]}...")
    
    # Get hierarchical context (current + parents)
    hierarchical_context = await api.context_get(depth=3)
    print(f"\nHierarchical context ({len(hierarchical_context)} blobs):")
    for blob in hierarchical_context:
        print(f"  - {blob.content_type}: {blob.content[:50]}...")
    
    # 4. Semantic search
    print("\n4. Semantic Search")
    
    # Search for authentication-related context
    search_results = await api.context_search("authentication security")
    print(f"Search results for 'authentication security' ({len(search_results)} results):")
    for blob in search_results:
        print(f"  - Score: {blob.metadata.get('search_score', 'N/A')}")
        print(f"    Content: {blob.content[:60]}...")
        print(f"    Tags: {', '.join(blob.tags)}")
    
    # Search within specific scope
    api_results = await api.context_search("API endpoints", scope="/project/src")
    print(f"\nAPI-related results in /project/src ({len(api_results)} results):")
    for blob in api_results:
        print(f"  - {blob.content[:50]}...")
    
    # 5. Directory listing
    print("\n5. Directory Listing")
    
    await api.context_cd("/project")
    paths = await api.context_ls()
    print(f"Paths in /project: {paths}")
    
    await api.context_cd("/")
    all_paths = await api.context_ls()
    print(f"All top-level paths: {all_paths}")
    
    # 6. Context updates
    print("\n6. Context Updates")
    
    # Update existing context
    success = await api.context_update(blob_id, {
        "content": "Enhanced authentication module with 2FA support and rate limiting",
        "tags": ["auth", "jwt", "security", "endpoints", "2fa", "rate-limiting"],
        "metadata": {
            "module": "auth.py", 
            "security_level": "high",
            "features": ["2fa", "rate_limiting"]
        }
    })
    print(f"Updated auth context: {success}")
    
    # Verify update
    updated_context = await api.context_get(depth=1)
    for blob in updated_context:
        if blob.id == blob_id:
            print(f"Updated content: {blob.content[:60]}...")
            print(f"New tags: {', '.join(blob.tags)}")
    
    print("\n=== Example Complete ===")


async def subscription_example():
    """Example of context subscription for real-time updates."""
    
    print("\n=== Subscription Example ===")
    
    api = ContextAPI()
    
    # Define callback for context changes
    def on_context_change(event):
        print(f"Context change detected:")
        print(f"  Path: {event.path}")
        print(f"  Type: {event.change_type}")
        print(f"  Blob ID: {event.blob_id}")
        print(f"  Timestamp: {event.timestamp}")
    
    # Subscribe to changes in project directory
    sub_id = await api.context_subscribe(
        "/project",
        callback=on_context_change,
        filters={"change_type": ["create", "update"]}
    )
    print(f"Subscribed to /project changes: {sub_id}")
    
    # Simulate context changes
    await asyncio.sleep(1)
    
    new_context = ContextBlob(
        content="New feature implementation notes",
        content_type="documentation",
        tags=["feature", "implementation"],
        metadata={"status": "draft"}
    )
    
    await api.context_push("/project/new_feature", new_context)
    await asyncio.sleep(1)  # Allow subscription callback to fire
    
    # Unsubscribe
    success = await api.context_unsubscribe(sub_id)
    print(f"Unsubscribed: {success}")


if __name__ == "__main__":
    # Run basic example
    asyncio.run(basic_example())
    
    # Run subscription example
    asyncio.run(subscription_example())