🔍 Code Extractor

class ManualRelationshipManager

Maturity: 53

A class that manages manually defined database relationships with persistent JSON storage, allowing users to add, retrieve, update, and remove relationship definitions between database tables.

File:
/tf/active/vicechatdev/full_smartstat/manual_relationships.py
Lines:
16 - 153
Complexity:
moderate

Purpose

This class provides a complete management system for manually defined database relationships. It handles persistent storage of relationships in JSON format, supports CRUD operations (create, read, update, delete), and provides querying capabilities to retrieve relationships by table or specific column pairs. The class is designed for scenarios where database relationships need to be explicitly defined and maintained outside of the database schema itself, such as for documentation, data lineage tracking, or relationship discovery systems.

Source Code

class ManualRelationshipManager:
    """Manages manually defined database relationships with persistent storage."""
    
    def __init__(self, storage_file: str = "manual_relationships.json"):
        """Initialize the manager with a storage file path."""
        self.storage_file = storage_file
        self.relationships = self._load_relationships()
    
    def _load_relationships(self) -> List[Dict[str, Any]]:
        """Load relationships from the storage file."""
        if not os.path.exists(self.storage_file):
            return []
        
        try:
            with open(self.storage_file, 'r') as f:
                data = json.load(f)
                return data.get('relationships', [])
        except (json.JSONDecodeError, FileNotFoundError) as e:
            logger.warning(f"Could not load manual relationships: {e}")
            return []
    
    def _save_relationships(self):
        """Save relationships to the storage file."""
        data = {
            'relationships': self.relationships,
            'last_updated': datetime.now().isoformat(),
            'version': '1.0'
        }
        
        try:
            with open(self.storage_file, 'w') as f:
                json.dump(data, f, indent=2)
            logger.info(f"Saved {len(self.relationships)} manual relationships to {self.storage_file}")
        except Exception as e:
            logger.error(f"Could not save manual relationships: {e}")
    
    def add_relationship(self, 
                        from_table: str, 
                        from_column: str, 
                        to_table: str, 
                        to_column: str,
                        confidence: float = 1.0,
                        description: str = None,
                        created_by: str = "manual") -> bool:
        """Add a manual relationship."""
        
        # Check if relationship already exists
        existing = self.get_relationship(from_table, from_column, to_table, to_column)
        if existing:
            logger.info(f"Relationship {from_table}.{from_column} → {to_table}.{to_column} already exists")
            return False
        
        relationship = {
            'from_table': from_table,
            'from_column': from_column,
            'to_table': to_table,
            'to_column': to_column,
            'confidence': confidence,
            'detection_method': 'manual',
            'created_at': datetime.now().isoformat(),
            'created_by': created_by,
            'description': description or f"Manual relationship: {from_table}.{from_column} → {to_table}.{to_column}"
        }
        
        self.relationships.append(relationship)
        self._save_relationships()
        
        logger.info(f"Added manual relationship: {from_table}.{from_column} → {to_table}.{to_column}")
        return True
    
    def get_relationship(self, from_table: str, from_column: str, to_table: str, to_column: str) -> Optional[Dict[str, Any]]:
        """Get a specific relationship if it exists."""
        for rel in self.relationships:
            if (rel['from_table'].lower() == from_table.lower() and 
                rel['from_column'].lower() == from_column.lower() and
                rel['to_table'].lower() == to_table.lower() and
                rel['to_column'].lower() == to_column.lower()):
                return rel
        return None
    
    def get_all_relationships(self) -> List[Dict[str, Any]]:
        """Get all manual relationships."""
        return self.relationships.copy()
    
    def get_relationships_for_table(self, table_name: str) -> List[Dict[str, Any]]:
        """Get all relationships where the table is either the source or target."""
        table_lower = table_name.lower()
        return [rel for rel in self.relationships 
                if rel['from_table'].lower() == table_lower or rel['to_table'].lower() == table_lower]
    
    def remove_relationship(self, from_table: str, from_column: str, to_table: str, to_column: str) -> bool:
        """Remove a manual relationship."""
        original_count = len(self.relationships)
        
        self.relationships = [rel for rel in self.relationships 
                            if not (rel['from_table'].lower() == from_table.lower() and 
                                   rel['from_column'].lower() == from_column.lower() and
                                   rel['to_table'].lower() == to_table.lower() and
                                   rel['to_column'].lower() == to_column.lower())]
        
        if len(self.relationships) < original_count:
            self._save_relationships()
            logger.info(f"Removed manual relationship: {from_table}.{from_column} → {to_table}.{to_column}")
            return True
        
        return False
    
    def clear_all_relationships(self) -> int:
        """Clear all manual relationships."""
        count = len(self.relationships)
        self.relationships = []
        self._save_relationships()
        logger.info(f"Cleared {count} manual relationships")
        return count
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics about manual relationships."""
        if not self.relationships:
            return {
                'total_relationships': 0,
                'tables_involved': 0,
                'average_confidence': 0.0
            }
        
        tables = set()
        total_confidence = 0
        
        for rel in self.relationships:
            tables.add(rel['from_table'])
            tables.add(rel['to_table'])
            total_confidence += rel.get('confidence', 1.0)
        
        return {
            'total_relationships': len(self.relationships),
            'tables_involved': len(tables),
            'average_confidence': total_confidence / len(self.relationships),
            'last_updated': max([rel.get('created_at', '') for rel in self.relationships], default='')
        }

Parameters

Name Type Default Kind
bases - -

Parameter Details

storage_file: Path to the JSON file where relationships will be persisted. Defaults to 'manual_relationships.json' in the current directory. The file will be created if it doesn't exist, and loaded on initialization if it does exist.

Return Value

Instantiation returns a ManualRelationshipManager object with loaded relationships from the storage file. Key method returns: add_relationship() returns bool (True if added, False if already exists), get_relationship() returns Optional[Dict] (relationship dict or None), get_all_relationships() returns List[Dict] of all relationships, remove_relationship() returns bool (True if removed, False if not found), clear_all_relationships() returns int (count of cleared relationships), get_statistics() returns Dict with statistics about stored relationships.

Class Interface

Methods

__init__(self, storage_file: str = 'manual_relationships.json')

Purpose: Initialize the manager with a storage file path and load existing relationships

Parameters:

  • storage_file: Path to JSON file for persistent storage, defaults to 'manual_relationships.json'

Returns: None (constructor)

_load_relationships(self) -> List[Dict[str, Any]]

Purpose: Private method to load relationships from the storage file

Returns: List of relationship dictionaries, empty list if file doesn't exist or on error

_save_relationships(self)

Purpose: Private method to save current relationships to the storage file with metadata

Returns: None (saves to file as side effect)

add_relationship(self, from_table: str, from_column: str, to_table: str, to_column: str, confidence: float = 1.0, description: str = None, created_by: str = 'manual') -> bool

Purpose: Add a new manual relationship between two table columns

Parameters:

  • from_table: Source table name
  • from_column: Source column name
  • to_table: Target table name
  • to_column: Target column name
  • confidence: Confidence score for the relationship (0.0 to 1.0), defaults to 1.0
  • description: Optional description of the relationship, auto-generated if not provided
  • created_by: Identifier of who created the relationship, defaults to 'manual'

Returns: True if relationship was added, False if it already exists

get_relationship(self, from_table: str, from_column: str, to_table: str, to_column: str) -> Optional[Dict[str, Any]]

Purpose: Retrieve a specific relationship by its source and target identifiers

Parameters:

  • from_table: Source table name (case-insensitive)
  • from_column: Source column name (case-insensitive)
  • to_table: Target table name (case-insensitive)
  • to_column: Target column name (case-insensitive)

Returns: Dictionary containing relationship details if found, None otherwise

get_all_relationships(self) -> List[Dict[str, Any]]

Purpose: Retrieve all stored relationships

Returns: Copy of the list containing all relationship dictionaries

get_relationships_for_table(self, table_name: str) -> List[Dict[str, Any]]

Purpose: Get all relationships where the specified table is either source or target

Parameters:

  • table_name: Name of the table to search for (case-insensitive)

Returns: List of relationship dictionaries involving the specified table

remove_relationship(self, from_table: str, from_column: str, to_table: str, to_column: str) -> bool

Purpose: Remove a specific relationship from storage

Parameters:

  • from_table: Source table name (case-insensitive)
  • from_column: Source column name (case-insensitive)
  • to_table: Target table name (case-insensitive)
  • to_column: Target column name (case-insensitive)

Returns: True if relationship was found and removed, False if not found

clear_all_relationships(self) -> int

Purpose: Remove all relationships from storage

Returns: Count of relationships that were cleared

get_statistics(self) -> Dict[str, Any]

Purpose: Calculate and return statistics about stored relationships

Returns: Dictionary with keys: total_relationships (int), tables_involved (int), average_confidence (float), last_updated (str)

Attributes

Name Type Description Scope
storage_file str Path to the JSON file used for persistent storage of relationships instance
relationships List[Dict[str, Any]] In-memory list of relationship dictionaries loaded from storage file. Each dict contains: from_table, from_column, to_table, to_column, confidence, detection_method, created_at, created_by, description instance

Dependencies

  • json
  • os
  • datetime
  • typing
  • logging

Required Imports

import json
import os
from datetime import datetime
from typing import List, Dict, Any, Optional
import logging

Usage Example

import json
import os
from datetime import datetime
from typing import List, Dict, Any, Optional
import logging

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Instantiate the manager
manager = ManualRelationshipManager(storage_file='my_relationships.json')

# Add a relationship
success = manager.add_relationship(
    from_table='orders',
    from_column='customer_id',
    to_table='customers',
    to_column='id',
    confidence=1.0,
    description='Orders to customers foreign key',
    created_by='admin'
)

# Get a specific relationship
rel = manager.get_relationship('orders', 'customer_id', 'customers', 'id')
if rel:
    print(f"Found relationship: {rel['description']}")

# Get all relationships for a table
table_rels = manager.get_relationships_for_table('orders')
print(f"Found {len(table_rels)} relationships for orders table")

# Get all relationships
all_rels = manager.get_all_relationships()

# Get statistics
stats = manager.get_statistics()
print(f"Total relationships: {stats['total_relationships']}")

# Remove a relationship
removed = manager.remove_relationship('orders', 'customer_id', 'customers', 'id')

# Clear all relationships
count = manager.clear_all_relationships()

Best Practices

  • Always ensure a logger is configured before instantiating the class to avoid NameError
  • The storage file is automatically saved after each modification (add, remove, clear), so no explicit save call is needed
  • Relationship comparisons are case-insensitive for table and column names
  • The class maintains an in-memory copy of relationships that is synchronized with the file system
  • Use get_relationship() before add_relationship() if you need to check for duplicates explicitly
  • The storage file includes metadata (last_updated, version) in addition to relationships
  • All relationships have a confidence score (defaults to 1.0 for manual entries)
  • The created_at timestamp is automatically set to the current time when adding relationships
  • Handle file I/O exceptions gracefully - the class logs warnings/errors but continues operation
  • Use get_all_relationships() with .copy() to avoid modifying the internal state accidentally
  • The class is not thread-safe - use external locking if accessing from multiple threads

Tags

database relationships persistence json-storage data-management crud metadata foreign-keys schema-management relationship-tracking

Similar Components

AI-powered semantic similarity - components with related functionality:

  • function get_manual_relationship_manager 63.9% similar

    Returns a singleton instance of the ManualRelationshipManager class, creating it on first access using lazy initialization.

    From: /tf/active/vicechatdev/full_smartstat/manual_relationships.py

  • function add_manual_relationship 63.4% similar

    Flask route handler that adds a manually specified database relationship to the schema and clears the schema cache to force reload with the new relationship.

    From: /tf/active/vicechatdev/full_smartstat/app.py

  • class Neo4jManager 58.7% similar

    A manager class that provides a high-level interface for interacting with Neo4j graph databases, handling connections, queries, node creation, and relationship management.

    From: /tf/active/vicechatdev/QA_updater/knowledge_store/neo4j_manager.py

  • class DatabaseSchema_v1 53.3% similar

    A dataclass that represents database schema information, including table categories, relationships, and system architecture. Provides functionality to load schema from JSON files.

    From: /tf/active/vicechatdev/smartstat/sql_query_generator.py

  • class RelTypes 53.2% similar

    A constants class that defines string literals representing relationship types used in a graph database (Neo4j) for document management and approval workflows.

    From: /tf/active/vicechatdev/CDocs/db/schema_manager.py
← Back to Browse