CleverRDFlib/docs/observers.md

Observers API

CleverRDFLib implements the Observer pattern to allow monitoring and reacting to events during ontology loading and class hierarchy building. This enables you to track progress, collect errors, log events, and implement custom behavior.

Purpose

Observers serve the following purposes:

• Event monitoring: Track loading and build events
• Error collection: Collect errors and inconsistencies
• Progress tracking: Monitor loading progress
• Custom behavior: Implement custom logic based on events
• Logging: Log events for debugging and auditing

Observer Types

The framework provides two types of observers:

1. Loading Observers: Monitor ontology loading events
2. Class Hierarchy Build Observers: Monitor class hierarchy building events

Loading Observers

Loading observers monitor events during ontology loading:

BaseLoadingObserver

Base class for loading observers with default no-op implementations:

from cleverrdf_lib.observers.base_observer import BaseLoadingObserver

class MyLoadingObserver(BaseLoadingObserver):
    """Custom loading observer."""
    
    def on_loading_started(self, uri: str) -> None:
        """Called when loading starts."""
        print(f"Loading started: {uri}")
    
    def on_loading_completed(self, loaded_count: int) -> None:
        """Called when loading completes."""
        print(f"Loading completed: {loaded_count} ontologies loaded")
    
    def on_loading_failed(self, uri: str, error: Exception) -> None:
        """Called when loading fails."""
        print(f"Loading failed: {uri} - {error}")

Registering Loading Observers

from cleverrdf_lib import OntologyLoader

# Create observer
observer = MyLoadingObserver()

# Register with loader
loader = OntologyLoader(loading_observers=[observer])

# Load ontology (observer will be notified)
result = loader.load_ontology("ontology.ttl")

Multiple Observers

You can register multiple observers:

observer1 = MyLoadingObserver()
observer2 = AnotherLoadingObserver()

loader = OntologyLoader(loading_observers=[observer1, observer2])
result = loader.load_ontology("ontology.ttl")

CollectingObserver

Built-in observer that collects loading events:

from cleverrdf_lib.observers.base_observer import CollectingObserver

# Create collecting observer
collector = CollectingObserver()

loader = OntologyLoader(loading_observers=[collector])
result = loader.load_ontology("ontology.ttl")

# Access collected events
print(f"Started URI: {collector.started_uri}")
print(f"Completed count: {collector.completed_count}")
print(f"Failed URIs: {collector.failed_uris}")

CompositeObserver

The framework automatically uses CompositeObserver to manage multiple observers:

from cleverrdf_lib.observers.base_observer import CompositeObserver

# Create composite observer
composite = CompositeObserver([observer1, observer2, observer3])

loader = OntologyLoader(loading_observers=[composite])
result = loader.load_ontology("ontology.ttl")

Class Hierarchy Build Observers

Build observers monitor events during class hierarchy building:

BaseClassHierarchyBuildObserver

Base class for build observers with default no-op implementations:

from cleverrdf_lib.observers.base_hierarchy_oberver import BaseClassHierarchyBuildObserver

class MyBuildObserver(BaseClassHierarchyBuildObserver):
    """Custom build observer."""
    
    def on_build_started(self, iri: str) -> None:
        """Called when build starts."""
        print(f"Build started: {iri}")
    
    def on_build_completed(self, iri: str) -> None:
        """Called when build completes."""
        print(f"Build completed: {iri}")
    
    def on_build_failed(self, iri: str, elem_iri: str, error: Exception) -> None:
        """Called when build fails."""
        print(f"Build failed: {iri} - Element: {elem_iri} - Error: {error}")
    
    def on_build_inconsistency(self, iri: str, elem_iri: str, msg: str) -> None:
        """Called when an inconsistency is detected."""
        print(f"Inconsistency: {iri} - Element: {elem_iri} - Message: {msg}")

Registering Build Observers

from cleverrdf_lib import OntologyLoader

# Create observer
build_observer = MyBuildObserver()

# Register with loader
loader = OntologyLoader(hierarchy_build_observers=[build_observer])

# Load ontology (observer will be notified during hierarchy building)
result = loader.load_ontology("ontology.ttl")
hierarchy = result.class_hierarchy  # Build triggers observer notifications

Setting Observers on LoadingResult

You can also set build observers on the LoadingResult:

from cleverrdf_lib import OntologyLoader

loader = OntologyLoader()
result = loader.load_ontology("ontology.ttl")

# Set observer before accessing hierarchy
build_observer = MyBuildObserver()
result.set_class_hierarchy_build_observer(build_observer)

# Or set multiple observers
result.set_class_hierarchy_build_observers([observer1, observer2])

# Access hierarchy (observer will be notified)
hierarchy = result.class_hierarchy

CompositeClassHierarchyBuildObserver

The framework uses CompositeClassHierarchyBuildObserver to manage multiple build observers:

from cleverrdf_lib.observers.base_hierarchy_oberver import CompositeClassHierarchyBuildObserver

# Create composite observer
composite = CompositeClassHierarchyBuildObserver([observer1, observer2])

loader = OntologyLoader(hierarchy_build_observers=[composite])
result = loader.load_ontology("ontology.ttl")

Observer Events

Loading Events

Event	Method	Parameters	Description
Loading Started	on_loading_started	uri: str	Called when loading starts
Loading Completed	on_loading_completed	loaded_count: int	Called when loading completes
Loading Failed	on_loading_failed	uri: str, error: Exception	Called when loading fails

Build Events

Event	Method	Parameters	Description
Build Started	on_build_started	iri: str	Called when build starts
Build Completed	on_build_completed	iri: str	Called when build completes
Build Failed	on_build_failed	iri: str, elem_iri: str, error: Exception	Called when build fails
Build Inconsistency	on_build_inconsistency	iri: str, elem_iri: str, msg: str	Called when inconsistency detected

Example: Error Collector

Collect all errors during loading and building:

from cleverrdf_lib.observers.base_observer import BaseLoadingObserver
from cleverrdf_lib.observers.base_hierarchy_oberver import BaseClassHierarchyBuildObserver

class ErrorCollector(BaseLoadingObserver, BaseClassHierarchyBuildObserver):
    """Collects all errors and inconsistencies."""
    
    def __init__(self):
        self.loading_errors = []
        self.build_errors = []
        self.inconsistencies = []
    
    def on_loading_failed(self, uri: str, error: Exception) -> None:
        """Collect loading errors."""
        self.loading_errors.append({
            "uri": uri,
            "error": error
        })
    
    def on_build_failed(self, iri: str, elem_iri: str, error: Exception) -> None:
        """Collect build errors."""
        self.build_errors.append({
            "ontology_iri": iri,
            "element_iri": elem_iri,
            "error": error
        })
    
    def on_build_inconsistency(self, iri: str, elem_iri: str, msg: str) -> None:
        """Collect inconsistencies."""
        self.inconsistencies.append({
            "ontology_iri": iri,
            "element_iri": elem_iri,
            "message": msg
        })

# Use the collector
from cleverrdf_lib import OntologyLoader

collector = ErrorCollector()
loader = OntologyLoader(
    loading_observers=[collector],
    hierarchy_build_observers=[collector]
)

result = loader.load_ontology("ontology.ttl")
hierarchy = result.class_hierarchy

# Check collected errors
if collector.loading_errors:
    print("Loading errors:")
    for error in collector.loading_errors:
        print(f"  {error['uri']}: {error['error']}")

if collector.build_errors:
    print("Build errors:")
    for error in collector.build_errors:
        print(f"  {error['element_iri']}: {error['error']}")

if collector.inconsistencies:
    print("Inconsistencies:")
    for inc in collector.inconsistencies:
        print(f"  {inc['element_iri']}: {inc['message']}")

Example: Progress Logger

Log progress during loading and building:

import logging
from cleverrdf_lib.observers.base_observer import BaseLoadingObserver
from cleverrdf_lib.observers.base_hierarchy_oberver import BaseClassHierarchyBuildObserver
from cleverrdf_lib import OntologyLoader

class ProgressLogger(BaseLoadingObserver, BaseClassHierarchyBuildObserver):
    """Logs progress during loading and building."""
    
    def __init__(self):
        self.logger = logging.getLogger("ProgressLogger")
    
    def on_loading_started(self, uri: str) -> None:
        """Log loading start."""
        self.logger.info(f"Loading started: {uri}")
    
    def on_loading_completed(self, loaded_count: int) -> None:
        """Log loading completion."""
        self.logger.info(f"Loading completed: {loaded_count} ontologies loaded")
    
    def on_build_started(self, iri: str) -> None:
        """Log build start."""
        self.logger.info(f"Build started: {iri}")
    
    def on_build_completed(self, iri: str) -> None:
        """Log build completion."""
        self.logger.info(f"Build completed: {iri}")

# Use the logger
logger = ProgressLogger()
loader = OntologyLoader(
    loading_observers=[logger],
    hierarchy_build_observers=[logger]
)

result = loader.load_ontology("ontology.ttl")

Dynamic Observer Management

You can add and remove observers dynamically:

from cleverrdf_lib.observers.base_observer import CompositeObserver

# Create composite observer
composite = CompositeObserver()

# Add observers
observer1 = MyLoadingObserver()
observer2 = AnotherLoadingObserver()

composite.add_observer(observer1)
composite.add_observer(observer2)

# Remove observer
composite.remove_observer(observer1)

# Use with loader
loader = OntologyLoader(loading_observers=[composite])

Best Practices

1. Use base classes: Extend BaseLoadingObserver or BaseClassHierarchyBuildObserver
2. Handle exceptions: Don't let observer methods raise exceptions
3. Keep observers lightweight: Avoid expensive operations in observer methods
4. Use composite observers: Use composite observers for multiple observers
5. Test observers: Test observers with various scenarios

Observer Interfaces

LoadingObserver

class LoadingObserver(ABC):
    """Interface for loading observers."""
    
    @abstractmethod
    def on_loading_started(self, uri: str) -> None:
        """Called when loading starts."""
        pass
    
    @abstractmethod
    def on_loading_completed(self, loaded_count: int) -> None:
        """Called when loading completes."""
        pass
    
    @abstractmethod
    def on_loading_failed(self, uri: str, error: Exception) -> None:
        """Called when loading fails."""
        pass

ClassHierarchyBuildObserver

class ClassHierarchyBuildObserver:
    """Interface for build observers."""
    
    def on_build_started(self, iri: str) -> None:
        """Called when build starts."""
        pass
    
    def on_build_completed(self, iri: str) -> None:
        """Called when build completes."""
        pass
    
    def on_build_failed(self, iri: str, elem_iri: str, error: Exception) -> None:
        """Called when build fails."""
        pass
    
    def on_build_inconsistency(self, iri: str, elem_iri: str, msg: str) -> None:
        """Called when inconsistency is detected."""
        pass

Next Steps

• Learn about Error Handling
• Explore Class Hierarchy Navigation
• Understand Loading Result API