object_registry

langroid/utils/object_registry.py

ObjectRegistry

A global registry to hold id -> object mappings.

add(obj) classmethod

Adds an object to the registry, returning the object's ID.

Source code in langroid/utils/object_registry.py

@classmethod
def add(cls, obj: ObjWithId) -> str:
    """Adds an object to the registry, returning the object's ID."""
    object_id = obj.id() if callable(obj.id) else obj.id
    cls.registry[object_id] = obj
    return object_id

get(obj_id) classmethod

Retrieves an object by ID if it still exists.

Source code in langroid/utils/object_registry.py

@classmethod
def get(cls, obj_id: str) -> Optional[ObjWithId]:
    """Retrieves an object by ID if it still exists."""
    return cls.registry.get(obj_id)

register_object(obj) classmethod

Registers an object in the registry, returning the object's ID.

Source code in langroid/utils/object_registry.py

@classmethod
def register_object(cls, obj: ObjWithId) -> str:
    """Registers an object in the registry, returning the object's ID."""
    return cls.add(obj)

remove(obj_id) classmethod

Removes an object from the registry.

Source code in langroid/utils/object_registry.py

@classmethod
def remove(cls, obj_id: str) -> None:
    """Removes an object from the registry."""
    if obj_id in cls.registry:
        del cls.registry[obj_id]

cleanup() classmethod

Cleans up the registry by removing entries where the object is None.

Source code in langroid/utils/object_registry.py

@classmethod
def cleanup(cls) -> None:
    """Cleans up the registry by removing entries where the object is None."""
    to_remove = [key for key, value in cls.registry.items() if value is None]
    for key in to_remove:
        del cls.registry[key]

new_id() staticmethod

Generates a new unique ID.

Source code in langroid/utils/object_registry.py

@staticmethod
def new_id() -> str:
    """Generates a new unique ID."""
    return str(uuid4())

scheduled_cleanup(interval=600)

Periodically cleans up the global registry every 'interval' seconds.

Source code in langroid/utils/object_registry.py

def scheduled_cleanup(interval: int = 600) -> None:
    """Periodically cleans up the global registry every 'interval' seconds."""
    while True:
        ObjectRegistry.cleanup()
        time.sleep(interval)