Skip to content

PetFactory

Experimental Feature

This component is part of the experimental Remote Injection system. Use with caution in production environments.

labchain.container.persistent.pet_factory.PetFactory

Bases: BaseFactory[TypePlugable]

Persistent factory with version control and lazy loading.

This factory extends BaseFactory to provide automatic class persistence and retrieval from storage. It maintains local version tracking and can automatically fetch classes from storage when they're not in memory.

Key Features:

- Automatic hash computation on registration
- Lazy loading from storage when class not in memory
- Version-specific class retrieval for pipeline reconstruction
- Bulk push operation for all registered classes
- Integration with PetClassManager for storage operations
Usage 
from labchain.storage import S3Storage
from labchain.container.persistent import PetClassManager, PetFactory

storage = S3Storage(bucket="my-models")
manager = PetClassManager(storage)
factory = PetFactory(manager)

# Register classes
factory["MyFilter"] = MyFilter
factory["AnotherFilter"] = AnotherFilter

# Push to storage
factory.push_all()

# Later, on another machine
# Class auto-loads if not in memory
cls = factory["MyFilter"]

# Get specific version
old_version = factory.get_version("MyFilter", "abc123...")

Attributes:

Name Type Description
manager PetClassManager

Manager for storage operations.
version_control Dict[str, str]

Maps class names to their current hashes.

Methods:

Name Description
__setitem__

str, value: Type[TypePlugable]) -> None: Register class and compute its hash.
__getitem__

str) -> Type[TypePlugable]: Get class, with automatic storage fallback.
get

str, default: Optional[Type[TypePlugable]] = None) -> Optional[Type[TypePlugable]]: Get class with default fallback.
get_version

str, code_hash: str) -> Type[TypePlugable]: Get specific version by hash.
push_all

Push all registered classes to storage.
Note

This factory is designed to work seamlessly with PetContainer and BasePlugin.build_from_dump() for full pipeline portability.

Source code in labchain/container/persistent/pet_factory.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
class PetFactory(BaseFactory[TypePlugable]):
    """
    Persistent factory with version control and lazy loading.

    This factory extends BaseFactory to provide automatic class persistence
    and retrieval from storage. It maintains local version tracking and
    can automatically fetch classes from storage when they're not in memory.

    Key Features:

        - Automatic hash computation on registration
        - Lazy loading from storage when class not in memory
        - Version-specific class retrieval for pipeline reconstruction
        - Bulk push operation for all registered classes
        - Integration with PetClassManager for storage operations

    Usage:
        ```python
        from labchain.storage import S3Storage
        from labchain.container.persistent import PetClassManager, PetFactory

        storage = S3Storage(bucket="my-models")
        manager = PetClassManager(storage)
        factory = PetFactory(manager)

        # Register classes
        factory["MyFilter"] = MyFilter
        factory["AnotherFilter"] = AnotherFilter

        # Push to storage
        factory.push_all()

        # Later, on another machine
        # Class auto-loads if not in memory
        cls = factory["MyFilter"]

        # Get specific version
        old_version = factory.get_version("MyFilter", "abc123...")
        ```

    Attributes:
        manager (PetClassManager): Manager for storage operations.
        version_control (Dict[str, str]): Maps class names to their current hashes.

    Methods:
        __setitem__(name: str, value: Type[TypePlugable]) -> None:
            Register class and compute its hash.

        __getitem__(name: str) -> Type[TypePlugable]:
            Get class, with automatic storage fallback.

        get(name: str, default: Optional[Type[TypePlugable]] = None) -> Optional[Type[TypePlugable]]:
            Get class with default fallback.

        get_version(name: str, code_hash: str) -> Type[TypePlugable]:
            Get specific version by hash.

        push_all() -> None:
            Push all registered classes to storage.

    Note:
        This factory is designed to work seamlessly with PetContainer
        and BasePlugin.build_from_dump() for full pipeline portability.
    """

    def __init__(self, class_manager: PetClassManager, fallback_factory: BaseFactory):
        """
        Initialize the PetFactory with a class manager.

        Args:
            class_manager (PetClassManager): Manager for storage operations.

        Example:
        ```python
            manager = PetClassManager(storage)
            factory = PetFactory(manager)
        ```
        """
        super().__init__()
        os.environ["TYPEGUARD_DISABLE"] = "1"
        self._manager = class_manager
        self._fallback_factory = fallback_factory
        self._version_control: Dict[str, str] = {}

    def __setitem__(self, name: str, value: Type[TypePlugable]) -> None:
        """
        Register class locally and compute its content hash.

        When a class is registered, this method:

            1. Computes the SHA-256 hash of the class source code
            2. Stores the hash in version_control for tracking
            3. Registers the class in the parent factory

        Args:
            name (str): Name to register the class under.
            value (Type[TypePlugable]): The class to register.

        Returns:
            None

        Example:
        ```python
            factory["MyFilter"] = MyFilter
            # Hash is automatically computed and tracked
        ```

        Note:
            The hash is computed immediately but the class is not pushed
            to storage until push_all() is called.
        """
        # code_hash = self._manager.get_class_hash(value)
        class_hash = cast(BasePlugin, value)._hash
        if class_hash is not None:
            self._version_control[name] = class_hash
            super().__setitem__(name, value)
        else:
            raise ValueError("Class must be a BasePlugin and should have a hash")

    def __getitem__(
        self, name: str, default: Type[TypePlugable] | None = None
    ) -> Type[TypePlugable]:
        """
        Get class with automatic lazy loading from storage.

        This method implements the following fallback chain:

            1. Try to get from local memory (parent factory)
            2. If not found, check storage for 'latest' version
            3. If found in storage, pull and register it
            4. If not found anywhere, raise AttributeError

        Args:
            name (str): Name of the class to retrieve.

        Returns:
            Type[TypePlugable]: The requested class.

        Raises:
            AttributeError: If class not found locally or in storage.

        Example:
        ```python
            # Class in memory - instant return
            cls = factory["MyFilter"]

            # Class not in memory - auto-pulls from storage
            cls = factory["RemoteFilter"]
        ```

        Note:
            Lazy loading is transparent - users don't need to explicitly
            call pull() before accessing classes.
        """
        try:
            return super().__getitem__(name)
        except AttributeError:
            # Not in memory, try to fetch 'latest' from storage
            remote_meta = self._manager._get_remote_latest_meta(name)

            if remote_meta:
                code_hash = remote_meta["hash"]
                rprint(
                    f"[yellow]Class {name} missing. Pulling latest: {code_hash[:8]}...[/yellow]"
                )

                # Pull and register dynamically
                recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)
                self._version_control[name] = code_hash
                super().__setitem__(name, recovered_cls)
                return recovered_cls

            if name in self._fallback_factory:
                return self._fallback_factory[name]

            raise AttributeError(
                f"Class '{name}' not found locally and no remote version exists."
            )

    def get(
        self, name: str, default: Type[TypePlugable] | None = None
    ) -> Type[TypePlugable] | None:
        """
        Get class with optional default, attempting lazy load from storage.

        This method overrides the parent's get() to add lazy loading capability.
        It provides a safe way to retrieve classes with a fallback value.

        Args:
            name (str): Name of the class to retrieve.
            default (Optional[Type[TypePlugable]]): Default value if not found.

        Returns:
            Optional[Type[TypePlugable]]: The class or default value.

        Example:
        ```python
            # With default
            cls = factory.get("MaybeExists", DefaultFilter)

            # Without default (returns None if not found)
            cls = factory.get("MaybeExists")
            if cls is None:
                print("Class not found")
        ```

        Note:
            Unlike __getitem__, this method returns the default value
            (or None) instead of raising an exception when the class
            is not found.
        """
        try:
            return self.__getitem__(name)
        except AttributeError:
            return default

    def get_version(self, name: str, code_hash: str) -> Type[TypePlugable]:
        """
        Get a specific version of a class by its hash.

        This method is essential for reconstructing old pipelines via
        BasePlugin.build_from_dump(). It allows loading exact versions
        of classes as they existed when a pipeline was saved.

        Args:
            name (str): Name of the class.
            code_hash (str): SHA-256 hash of the specific version.

        Returns:
            Type[TypePlugable]: The requested version of the class.

        Example:
        ```python
            # Save pipeline with current versions
            config = pipeline.item_dump()  # Includes version_hash

            # Later, reconstruct with exact versions
            old_filter = factory.get_version(
                "MyFilter",
                config["params"]["filter"]["version_hash"]
            )
        ```

        Note:
            This method does NOT register the fetched version in the
            factory's main registry, avoiding conflicts with the current
            'latest' version. It simply returns the requested class.
        """
        if name in self._foundry and self._version_control.get(name) == code_hash:
            return self._foundry[name]

        rprint(
            f"[cyan]Reconstructing specific version {code_hash[:8]}... for {name}[/cyan]"
        )
        recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)

        return recovered_cls

    def push_all(self) -> None:
        """
        Push all locally registered classes to storage.

        This method iterates through all classes in the factory and
        pushes each one to storage, updating their 'latest' pointers.
        This is the primary way to synchronize local classes with storage.

        Returns:
            None

        Example:
        ```python
            # Register multiple classes
            factory["Filter1"] = Filter1
            factory["Filter2"] = Filter2
            factory["Filter3"] = Filter3

            # Push all at once
            factory.push_all()
        ```

        Note:
            This operation is idempotent - pushing the same class
            multiple times without changes will not create duplicate
            storage entries (same hash = same storage path).
        """
        for name, cls in self._foundry.items():
            rprint(f"[blue]Publishing {name}...[/blue]")
            self._manager.push(cls)

__init__(class_manager, fallback_factory)

Initialize the PetFactory with a class manager.

Parameters:

Name Type Description Default
class_manager PetClassManager

Manager for storage operations.

 required

Example: 

    manager = PetClassManager(storage)
    factory = PetFactory(manager)

Source code in labchain/container/persistent/pet_factory.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(self, class_manager: PetClassManager, fallback_factory: BaseFactory):
    """
    Initialize the PetFactory with a class manager.

    Args:
        class_manager (PetClassManager): Manager for storage operations.

    Example:
    ```python
        manager = PetClassManager(storage)
        factory = PetFactory(manager)
    ```
    """
    super().__init__()
    os.environ["TYPEGUARD_DISABLE"] = "1"
    self._manager = class_manager
    self._fallback_factory = fallback_factory
    self._version_control: Dict[str, str] = {}

__setitem__(name, value)

Register class locally and compute its content hash.

When a class is registered, this method:

1. Computes the SHA-256 hash of the class source code
2. Stores the hash in version_control for tracking
3. Registers the class in the parent factory

Parameters:

Name Type Description Default
name str

Name to register the class under.

 required
value Type[TypePlugable]

The class to register.

 required

Returns:

Type Description
None

None

Example: 

    factory["MyFilter"] = MyFilter
    # Hash is automatically computed and tracked

Note

The hash is computed immediately but the class is not pushed to storage until push_all() is called.

Source code in labchain/container/persistent/pet_factory.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def __setitem__(self, name: str, value: Type[TypePlugable]) -> None:
    """
    Register class locally and compute its content hash.

    When a class is registered, this method:

        1. Computes the SHA-256 hash of the class source code
        2. Stores the hash in version_control for tracking
        3. Registers the class in the parent factory

    Args:
        name (str): Name to register the class under.
        value (Type[TypePlugable]): The class to register.

    Returns:
        None

    Example:
    ```python
        factory["MyFilter"] = MyFilter
        # Hash is automatically computed and tracked
    ```

    Note:
        The hash is computed immediately but the class is not pushed
        to storage until push_all() is called.
    """
    # code_hash = self._manager.get_class_hash(value)
    class_hash = cast(BasePlugin, value)._hash
    if class_hash is not None:
        self._version_control[name] = class_hash
        super().__setitem__(name, value)
    else:
        raise ValueError("Class must be a BasePlugin and should have a hash")

__getitem__(name, default=None)

Get class with automatic lazy loading from storage.

This method implements the following fallback chain:

1. Try to get from local memory (parent factory)
2. If not found, check storage for 'latest' version
3. If found in storage, pull and register it
4. If not found anywhere, raise AttributeError

Parameters:

Name Type Description Default
name str

Name of the class to retrieve.

 required

Returns:

Type Description
Type[TypePlugable]

Type[TypePlugable]: The requested class.

Raises:

Type Description
AttributeError

If class not found locally or in storage.

Example: 

    # Class in memory - instant return
    cls = factory["MyFilter"]

    # Class not in memory - auto-pulls from storage
    cls = factory["RemoteFilter"]

Note

Lazy loading is transparent - users don't need to explicitly call pull() before accessing classes.

Source code in labchain/container/persistent/pet_factory.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def __getitem__(
    self, name: str, default: Type[TypePlugable] | None = None
) -> Type[TypePlugable]:
    """
    Get class with automatic lazy loading from storage.

    This method implements the following fallback chain:

        1. Try to get from local memory (parent factory)
        2. If not found, check storage for 'latest' version
        3. If found in storage, pull and register it
        4. If not found anywhere, raise AttributeError

    Args:
        name (str): Name of the class to retrieve.

    Returns:
        Type[TypePlugable]: The requested class.

    Raises:
        AttributeError: If class not found locally or in storage.

    Example:
    ```python
        # Class in memory - instant return
        cls = factory["MyFilter"]

        # Class not in memory - auto-pulls from storage
        cls = factory["RemoteFilter"]
    ```

    Note:
        Lazy loading is transparent - users don't need to explicitly
        call pull() before accessing classes.
    """
    try:
        return super().__getitem__(name)
    except AttributeError:
        # Not in memory, try to fetch 'latest' from storage
        remote_meta = self._manager._get_remote_latest_meta(name)

        if remote_meta:
            code_hash = remote_meta["hash"]
            rprint(
                f"[yellow]Class {name} missing. Pulling latest: {code_hash[:8]}...[/yellow]"
            )

            # Pull and register dynamically
            recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)
            self._version_control[name] = code_hash
            super().__setitem__(name, recovered_cls)
            return recovered_cls

        if name in self._fallback_factory:
            return self._fallback_factory[name]

        raise AttributeError(
            f"Class '{name}' not found locally and no remote version exists."
        )

get(name, default=None)

Get class with optional default, attempting lazy load from storage.

This method overrides the parent's get() to add lazy loading capability. It provides a safe way to retrieve classes with a fallback value.

Parameters:

Name Type Description Default
name str

Name of the class to retrieve.

 required
default Optional[Type[TypePlugable]]

Default value if not found.

 None

Returns:

Type Description
Type[TypePlugable] | None

Optional[Type[TypePlugable]]: The class or default value.

Example: 

    # With default
    cls = factory.get("MaybeExists", DefaultFilter)

    # Without default (returns None if not found)
    cls = factory.get("MaybeExists")
    if cls is None:
        print("Class not found")

Note

Unlike getitem, this method returns the default value (or None) instead of raising an exception when the class is not found.

Source code in labchain/container/persistent/pet_factory.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def get(
    self, name: str, default: Type[TypePlugable] | None = None
) -> Type[TypePlugable] | None:
    """
    Get class with optional default, attempting lazy load from storage.

    This method overrides the parent's get() to add lazy loading capability.
    It provides a safe way to retrieve classes with a fallback value.

    Args:
        name (str): Name of the class to retrieve.
        default (Optional[Type[TypePlugable]]): Default value if not found.

    Returns:
        Optional[Type[TypePlugable]]: The class or default value.

    Example:
    ```python
        # With default
        cls = factory.get("MaybeExists", DefaultFilter)

        # Without default (returns None if not found)
        cls = factory.get("MaybeExists")
        if cls is None:
            print("Class not found")
    ```

    Note:
        Unlike __getitem__, this method returns the default value
        (or None) instead of raising an exception when the class
        is not found.
    """
    try:
        return self.__getitem__(name)
    except AttributeError:
        return default

get_version(name, code_hash)

Get a specific version of a class by its hash.

This method is essential for reconstructing old pipelines via BasePlugin.build_from_dump(). It allows loading exact versions of classes as they existed when a pipeline was saved.

Parameters:

Name Type Description Default
name str

Name of the class.

 required
code_hash str

SHA-256 hash of the specific version.

 required

Returns:

Type Description
Type[TypePlugable]

Type[TypePlugable]: The requested version of the class.

Example: 

    # Save pipeline with current versions
    config = pipeline.item_dump()  # Includes version_hash

    # Later, reconstruct with exact versions
    old_filter = factory.get_version(
        "MyFilter",
        config["params"]["filter"]["version_hash"]
    )

Note

This method does NOT register the fetched version in the factory's main registry, avoiding conflicts with the current 'latest' version. It simply returns the requested class.

Source code in labchain/container/persistent/pet_factory.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def get_version(self, name: str, code_hash: str) -> Type[TypePlugable]:
    """
    Get a specific version of a class by its hash.

    This method is essential for reconstructing old pipelines via
    BasePlugin.build_from_dump(). It allows loading exact versions
    of classes as they existed when a pipeline was saved.

    Args:
        name (str): Name of the class.
        code_hash (str): SHA-256 hash of the specific version.

    Returns:
        Type[TypePlugable]: The requested version of the class.

    Example:
    ```python
        # Save pipeline with current versions
        config = pipeline.item_dump()  # Includes version_hash

        # Later, reconstruct with exact versions
        old_filter = factory.get_version(
            "MyFilter",
            config["params"]["filter"]["version_hash"]
        )
    ```

    Note:
        This method does NOT register the fetched version in the
        factory's main registry, avoiding conflicts with the current
        'latest' version. It simply returns the requested class.
    """
    if name in self._foundry and self._version_control.get(name) == code_hash:
        return self._foundry[name]

    rprint(
        f"[cyan]Reconstructing specific version {code_hash[:8]}... for {name}[/cyan]"
    )
    recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)

    return recovered_cls

push_all()

Push all locally registered classes to storage.

This method iterates through all classes in the factory and pushes each one to storage, updating their 'latest' pointers. This is the primary way to synchronize local classes with storage.

Returns:

Type Description
None

None

Example: 

    # Register multiple classes
    factory["Filter1"] = Filter1
    factory["Filter2"] = Filter2
    factory["Filter3"] = Filter3

    # Push all at once
    factory.push_all()

Note

This operation is idempotent - pushing the same class multiple times without changes will not create duplicate storage entries (same hash = same storage path).

Source code in labchain/container/persistent/pet_factory.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def push_all(self) -> None:
    """
    Push all locally registered classes to storage.

    This method iterates through all classes in the factory and
    pushes each one to storage, updating their 'latest' pointers.
    This is the primary way to synchronize local classes with storage.

    Returns:
        None

    Example:
    ```python
        # Register multiple classes
        factory["Filter1"] = Filter1
        factory["Filter2"] = Filter2
        factory["Filter3"] = Filter3

        # Push all at once
        factory.push_all()
    ```

    Note:
        This operation is idempotent - pushing the same class
        multiple times without changes will not create duplicate
        storage entries (same hash = same storage path).
    """
    for name, cls in self._foundry.items():
        rprint(f"[blue]Publishing {name}...[/blue]")
        self._manager.push(cls)

Overview

PetFactory is a persistent-aware factory that extends BaseFactory with automatic version tracking and lazy loading from storage. It provides a seamless interface for working with persistent classes while maintaining full compatibility with the standard factory API.

Key Features

  • Automatic Hash Tracking: Computes and stores version hashes on registration
  • Lazy Loading: Automatically fetches classes from storage when not in memory
  • Version-Specific Retrieval: Load exact versions by hash for reproducibility
  • Bulk Operations: Push all registered classes with a single command
  • Transparent Fallback: Falls back to standard Container.pif for framework classes

Basic Usage

Initialize Factory

from labchain.storage import S3Storage
from labchain.container.persistent import PetClassManager, PetFactory
from labchain import Container

# Create storage and manager
storage = S3Storage(bucket="my-ml-models")
manager = PetClassManager(storage)

# Create factory (typically via Container)
factory = PetFactory(manager, Container.pif)

# Or use the pre-initialized Container.ppif
factory = Container.ppif

Register Classes

from labchain.base import BaseFilter

class MyFilter(BaseFilter):
    def __init__(self, threshold: float = 0.5):
        super().__init__(threshold=threshold)

    def predict(self, x):
        return x.value > self.threshold

# Register in factory (hash computed automatically)
factory["MyFilter"] = MyFilter

# Check version tracking
print(f"Tracked hash: {factory._version_control['MyFilter'][:8]}...")

Lazy Loading

# Class in memory - instant return
MyFilter = factory["MyFilter"]

# Class not in memory - auto-loads from storage
RemoteFilter = factory["RemoteFilter"]  # Pulls from storage automatically

Push to Storage

# Register multiple classes
factory["Filter1"] = Filter1
factory["Filter2"] = Filter2
factory["Filter3"] = Filter3

# Push all at once
factory.push_all()

Advanced Usage

Version-Specific Retrieval

# Save current version
config = my_pipeline.item_dump()
version_hash = config['params']['filter']['version_hash']

# Later, get exact version
MyFilterV1 = factory.get_version("MyFilter", version_hash)

# Use old version alongside new
MyFilterV2 = factory["MyFilter"]  # Latest

instance_v1 = MyFilterV1(threshold=0.3)
instance_v2 = MyFilterV2(threshold=0.5)

Safe Get with Default

# Try to get class, fallback to default
MyFilter = factory.get("MyOptionalFilter", default=DefaultFilter)

if MyFilter is None:
    print("Filter not found and no default provided")

Version Rollback Pattern

class VersionedFilterFactory:
    """Wrapper for version management."""

    def __init__(self, factory):
        self.factory = factory
        self.version_history = {}

    def register_with_tag(self, cls, tag):
        """Register class and tag the version."""
        self.factory[cls.__name__] = cls
        hash_value = self.factory._version_control[cls.__name__]
        self.version_history[cls.__name__] = {
            **self.version_history.get(cls.__name__, {}),
            tag: hash_value
        }
        return hash_value

    def get_by_tag(self, class_name, tag):
        """Get specific version by semantic tag."""
        hash_value = self.version_history[class_name][tag]
        return self.factory.get_version(class_name, hash_value)

    def rollback(self, class_name, tag):
        """Rollback to a previous version."""
        old_version = self.get_by_tag(class_name, tag)
        self.factory[class_name] = old_version
        return old_version

# Usage
vf = VersionedFilterFactory(Container.ppif)

# Version 1
class MyFilter(BaseFilter):
    def predict(self, x): return x * 1

vf.register_with_tag(MyFilter, "v1.0.0")

# Version 2
class MyFilter(BaseFilter):
    def predict(self, x): return x * 2

vf.register_with_tag(MyFilter, "v2.0.0")

# Rollback to v1
MyFilterV1 = vf.rollback("MyFilter", "v1.0.0")

Integration with Container

PetFactory is pre-initialized in Container.ppif: 

from labchain import Container

# Already available as Container.ppif
print(type(Container.ppif))  # <class 'PetFactory'>

# Register with decorator
@Container.bind(persist=True)
class MyFilter(BaseFilter):
    def predict(self, x):
        return x

# Automatically registered in Container.ppif
assert "MyFilter" in Container.ppif._foundry

# Push all persistent classes
Container.ppif.push_all()

Automatic vs Manual Registration

# Automatic (recommended)
@Container.bind(persist=True, auto_push=True)
class AutoFilter(BaseFilter):
    def predict(self, x): return x
# Automatically pushed to storage on registration

# Manual control
@Container.bind(persist=True)
class ManualFilter(BaseFilter):
    def predict(self, x): return x

# Explicitly push when ready
Container.ppif.push_all()

Fallback Behavior

PetFactory automatically falls back to Container.pif for framework classes: 

from labchain.pipeline import F3Pipeline

# F3Pipeline is a framework class (not user-defined)
# It's in Container.pif but NOT in Container.ppif

# PetFactory automatically falls back
pipeline_cls = Container.ppif["F3Pipeline"]  # Works! Falls back to Container.pif

# This enables build_from_dump to work seamlessly
config = pipeline.item_dump()
reconstructed = BasePlugin.build_from_dump(config, Container.ppif)
# Framework classes loaded from pif, user classes from ppif

Fallback Priority

def __getitem__(self, name):
    # Priority order:
    # 1. Check local memory (self._foundry)
    # 2. Check persistent storage (self.manager.pull)
    # 3. Check fallback factory (self._fallback_factory)
    # 4. Raise AttributeError
    pass

Working with build_from_dump

PetFactory integrates seamlessly with pipeline reconstruction: 

from labchain.base import BasePlugin
from labchain import Container

# Create and train pipeline
pipeline = F3Pipeline(filters=[MyCustomFilter(), StandardScalerPlugin()])
pipeline.fit(X_train, y_train)

# Serialize (includes version hashes)
config = pipeline.item_dump()

# Push custom classes
Container.ppif.push_all()

# Later, reconstruct (even on different machine)
reconstructed = BasePlugin.build_from_dump(config, Container.ppif)

# Uses get_version() internally to fetch exact versions

How build_from_dump Works

@staticmethod
def build_from_dump(dump_dict, factory):
    clazz_name = dump_dict["clazz"]
    v_hash = dump_dict.get("version_hash")

    if v_hash and hasattr(factory, "get_version"):
        # PetFactory has get_version - fetch specific version
        level_clazz = factory.get_version(clazz_name, v_hash)
    else:
        # Standard factory - get current version
        level_clazz = factory.get(clazz_name)

    # Recursively reconstruct nested plugins
    if "params" in dump_dict:
        params = {}
        for k, v in dump_dict["params"].items():
            if isinstance(v, dict) and "clazz" in v:
                params[k] = build_from_dump(v, factory)
            else:
                params[k] = v
        return level_clazz(**params)

    return level_clazz

Performance Optimization

Batch Registration

# ❌ Inefficient: Compute hash multiple times
for i in range(100):
    class DynamicFilter(BaseFilter):
        def predict(self, x): return x

    Container.ppif[f"Filter{i}"] = DynamicFilter

# ✅ Better: Pre-define classes
filters = {
    f"Filter{i}": create_filter(i)
    for i in range(100)
}

for name, cls in filters.items():
    Container.ppif[name] = cls

# Single push
Container.ppif.push_all()

Lazy Loading Benefits

# Only loads classes when actually needed
# No upfront cost for unused classes

def load_pipeline_lazy(config):
    """Reconstruct pipeline without loading all classes."""
    # build_from_dump only loads referenced classes
    pipeline = BasePlugin.build_from_dump(config, Container.ppif)

    # Other classes in storage remain unloaded
    return pipeline

Error Handling

Registration Errors

try:
    # Must have _hash attribute (set by @Container.bind(persist=True))
    Container.ppif["MyFilter"] = MyFilter
except ValueError as e:
    print(f"Registration failed: {e}")
    # Class must be decorated with @Container.bind(persist=True)
    # or have _hash attribute set manually

Loading Errors

try:
    cls = Container.ppif["MaybeExists"]
except AttributeError as e:
    print(f"Class not found: {e}")
    # Not in memory, storage, or fallback factory

# Safe alternative
cls = Container.ppif.get("MaybeExists", default=DefaultClass)

Version Errors

try:
    old_version = Container.ppif.get_version("MyFilter", "invalid_hash")
except ValueError as e:
    print(f"Version not found: {e}")
    # Hash doesn't exist in storage

Best Practices

1. Use Container.ppif via Decorator

# ✅ Recommended: Use decorator
@Container.bind(persist=True)
class MyFilter(BaseFilter):
    def predict(self, x): return x

# Automatically registered in Container.ppif

# ❌ Manual registration (more error-prone)
class MyFilter(BaseFilter):
    def predict(self, x): return x

MyFilter._hash = Container.pcm.get_class_hash(MyFilter)
Container.ppif["MyFilter"] = MyFilter

2. Push After Registration Batch

# ✅ Good: Register multiple, then push once
@Container.bind(persist=True)
class Filter1(BaseFilter): pass

@Container.bind(persist=True)
class Filter2(BaseFilter): pass

@Container.bind(persist=True)
class Filter3(BaseFilter): pass

Container.ppif.push_all()  # Single batch upload

# ❌ Bad: Push after each registration
@Container.bind(persist=True, auto_push=True)  # Uploads immediately
class Filter1(BaseFilter): pass

@Container.bind(persist=True, auto_push=True)  # Another upload
class Filter2(BaseFilter): pass
# Multiple network round-trips

3. Version Verification

def verify_version_consistency(factory, config):
    """Verify all version hashes in config exist in storage."""

    def extract_hashes(obj):
        """Recursively extract all version_hash values."""
        hashes = []
        if isinstance(obj, dict):
            if "version_hash" in obj:
                hashes.append((obj["clazz"], obj["version_hash"]))
            for v in obj.values():
                hashes.extend(extract_hashes(v))
        elif isinstance(obj, list):
            for item in obj:
                hashes.extend(extract_hashes(item))
        return hashes

    version_hashes = extract_hashes(config)

    for class_name, hash_value in version_hashes:
        try:
            factory.get_version(class_name, hash_value)
            print(f"✓ {class_name} {hash_value[:8]}... verified")
        except Exception as e:
            print(f"✗ {class_name} {hash_value[:8]}... MISSING")
            raise ValueError(f"Version verification failed: {e}")

    print("✓ All versions verified")

# Usage before deployment
verify_version_consistency(Container.ppif, pipeline_config)

Comparison: Standard vs Persistent Factory

Feature BaseFactory PetFactory
Registration factory[name] = cls factory[name] = cls (with hash)
Retrieval Memory only Memory → Storage → Fallback
Versioning None Automatic via hash
Storage Sync None push_all()
Lazy Loading No Yes
Version Pinning No get_version(name, hash)
Fallback No Yes (to Container.pif)

Troubleshooting

Issue: "Class must have a hash attribute"

Solution: Use @Container.bind(persist=True) decorator: 

# ❌ Wrong
class MyFilter(BaseFilter): pass
Container.ppif["MyFilter"] = MyFilter  # Error!

# ✅ Correct
@Container.bind(persist=True)
class MyFilter(BaseFilter): pass
# Hash automatically set

Issue: Lazy loading fails

Diagnostic: 

# Check if class exists in storage
meta = Container.pcm._get_remote_latest_meta("MyFilter")
print(f"Remote metadata: {meta}")

# Try explicit pull
if meta:
    try:
        cls = Container.pcm.pull("MyFilter", meta['hash'])
        print("✓ Manual pull successful")
    except Exception as e:
        print(f"✗ Pull failed: {e}")

Issue: Version not found

# List all available versions (if LocalStorage)
import os
storage_path = Container.storage.storage_path
class_dir = f"{storage_path}/plugins/MyFilter"

if os.path.exists(class_dir):
    files = os.listdir(class_dir)
    pkl_files = [f for f in files if f.endswith('.pkl')]
    print(f"Available versions: {pkl_files}")

See Also

API Reference

Bases: BaseFactory[TypePlugable]

Persistent factory with version control and lazy loading.

This factory extends BaseFactory to provide automatic class persistence and retrieval from storage. It maintains local version tracking and can automatically fetch classes from storage when they're not in memory.

Key Features:

- Automatic hash computation on registration
- Lazy loading from storage when class not in memory
- Version-specific class retrieval for pipeline reconstruction
- Bulk push operation for all registered classes
- Integration with PetClassManager for storage operations
Usage 
from labchain.storage import S3Storage
from labchain.container.persistent import PetClassManager, PetFactory

storage = S3Storage(bucket="my-models")
manager = PetClassManager(storage)
factory = PetFactory(manager)

# Register classes
factory["MyFilter"] = MyFilter
factory["AnotherFilter"] = AnotherFilter

# Push to storage
factory.push_all()

# Later, on another machine
# Class auto-loads if not in memory
cls = factory["MyFilter"]

# Get specific version
old_version = factory.get_version("MyFilter", "abc123...")

Attributes:

Name Type Description
manager PetClassManager

Manager for storage operations.
version_control Dict[str, str]

Maps class names to their current hashes.

Methods:

Name Description
__setitem__

str, value: Type[TypePlugable]) -> None: Register class and compute its hash.
__getitem__

str) -> Type[TypePlugable]: Get class, with automatic storage fallback.
get

str, default: Optional[Type[TypePlugable]] = None) -> Optional[Type[TypePlugable]]: Get class with default fallback.
get_version

str, code_hash: str) -> Type[TypePlugable]: Get specific version by hash.
push_all

Push all registered classes to storage.
Note

This factory is designed to work seamlessly with PetContainer and BasePlugin.build_from_dump() for full pipeline portability.

Source code in labchain/container/persistent/pet_factory.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
class PetFactory(BaseFactory[TypePlugable]):
    """
    Persistent factory with version control and lazy loading.

    This factory extends BaseFactory to provide automatic class persistence
    and retrieval from storage. It maintains local version tracking and
    can automatically fetch classes from storage when they're not in memory.

    Key Features:

        - Automatic hash computation on registration
        - Lazy loading from storage when class not in memory
        - Version-specific class retrieval for pipeline reconstruction
        - Bulk push operation for all registered classes
        - Integration with PetClassManager for storage operations

    Usage:
        ```python
        from labchain.storage import S3Storage
        from labchain.container.persistent import PetClassManager, PetFactory

        storage = S3Storage(bucket="my-models")
        manager = PetClassManager(storage)
        factory = PetFactory(manager)

        # Register classes
        factory["MyFilter"] = MyFilter
        factory["AnotherFilter"] = AnotherFilter

        # Push to storage
        factory.push_all()

        # Later, on another machine
        # Class auto-loads if not in memory
        cls = factory["MyFilter"]

        # Get specific version
        old_version = factory.get_version("MyFilter", "abc123...")
        ```

    Attributes:
        manager (PetClassManager): Manager for storage operations.
        version_control (Dict[str, str]): Maps class names to their current hashes.

    Methods:
        __setitem__(name: str, value: Type[TypePlugable]) -> None:
            Register class and compute its hash.

        __getitem__(name: str) -> Type[TypePlugable]:
            Get class, with automatic storage fallback.

        get(name: str, default: Optional[Type[TypePlugable]] = None) -> Optional[Type[TypePlugable]]:
            Get class with default fallback.

        get_version(name: str, code_hash: str) -> Type[TypePlugable]:
            Get specific version by hash.

        push_all() -> None:
            Push all registered classes to storage.

    Note:
        This factory is designed to work seamlessly with PetContainer
        and BasePlugin.build_from_dump() for full pipeline portability.
    """

    def __init__(self, class_manager: PetClassManager, fallback_factory: BaseFactory):
        """
        Initialize the PetFactory with a class manager.

        Args:
            class_manager (PetClassManager): Manager for storage operations.

        Example:
        ```python
            manager = PetClassManager(storage)
            factory = PetFactory(manager)
        ```
        """
        super().__init__()
        os.environ["TYPEGUARD_DISABLE"] = "1"
        self._manager = class_manager
        self._fallback_factory = fallback_factory
        self._version_control: Dict[str, str] = {}

    def __setitem__(self, name: str, value: Type[TypePlugable]) -> None:
        """
        Register class locally and compute its content hash.

        When a class is registered, this method:

            1. Computes the SHA-256 hash of the class source code
            2. Stores the hash in version_control for tracking
            3. Registers the class in the parent factory

        Args:
            name (str): Name to register the class under.
            value (Type[TypePlugable]): The class to register.

        Returns:
            None

        Example:
        ```python
            factory["MyFilter"] = MyFilter
            # Hash is automatically computed and tracked
        ```

        Note:
            The hash is computed immediately but the class is not pushed
            to storage until push_all() is called.
        """
        # code_hash = self._manager.get_class_hash(value)
        class_hash = cast(BasePlugin, value)._hash
        if class_hash is not None:
            self._version_control[name] = class_hash
            super().__setitem__(name, value)
        else:
            raise ValueError("Class must be a BasePlugin and should have a hash")

    def __getitem__(
        self, name: str, default: Type[TypePlugable] | None = None
    ) -> Type[TypePlugable]:
        """
        Get class with automatic lazy loading from storage.

        This method implements the following fallback chain:

            1. Try to get from local memory (parent factory)
            2. If not found, check storage for 'latest' version
            3. If found in storage, pull and register it
            4. If not found anywhere, raise AttributeError

        Args:
            name (str): Name of the class to retrieve.

        Returns:
            Type[TypePlugable]: The requested class.

        Raises:
            AttributeError: If class not found locally or in storage.

        Example:
        ```python
            # Class in memory - instant return
            cls = factory["MyFilter"]

            # Class not in memory - auto-pulls from storage
            cls = factory["RemoteFilter"]
        ```

        Note:
            Lazy loading is transparent - users don't need to explicitly
            call pull() before accessing classes.
        """
        try:
            return super().__getitem__(name)
        except AttributeError:
            # Not in memory, try to fetch 'latest' from storage
            remote_meta = self._manager._get_remote_latest_meta(name)

            if remote_meta:
                code_hash = remote_meta["hash"]
                rprint(
                    f"[yellow]Class {name} missing. Pulling latest: {code_hash[:8]}...[/yellow]"
                )

                # Pull and register dynamically
                recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)
                self._version_control[name] = code_hash
                super().__setitem__(name, recovered_cls)
                return recovered_cls

            if name in self._fallback_factory:
                return self._fallback_factory[name]

            raise AttributeError(
                f"Class '{name}' not found locally and no remote version exists."
            )

    def get(
        self, name: str, default: Type[TypePlugable] | None = None
    ) -> Type[TypePlugable] | None:
        """
        Get class with optional default, attempting lazy load from storage.

        This method overrides the parent's get() to add lazy loading capability.
        It provides a safe way to retrieve classes with a fallback value.

        Args:
            name (str): Name of the class to retrieve.
            default (Optional[Type[TypePlugable]]): Default value if not found.

        Returns:
            Optional[Type[TypePlugable]]: The class or default value.

        Example:
        ```python
            # With default
            cls = factory.get("MaybeExists", DefaultFilter)

            # Without default (returns None if not found)
            cls = factory.get("MaybeExists")
            if cls is None:
                print("Class not found")
        ```

        Note:
            Unlike __getitem__, this method returns the default value
            (or None) instead of raising an exception when the class
            is not found.
        """
        try:
            return self.__getitem__(name)
        except AttributeError:
            return default

    def get_version(self, name: str, code_hash: str) -> Type[TypePlugable]:
        """
        Get a specific version of a class by its hash.

        This method is essential for reconstructing old pipelines via
        BasePlugin.build_from_dump(). It allows loading exact versions
        of classes as they existed when a pipeline was saved.

        Args:
            name (str): Name of the class.
            code_hash (str): SHA-256 hash of the specific version.

        Returns:
            Type[TypePlugable]: The requested version of the class.

        Example:
        ```python
            # Save pipeline with current versions
            config = pipeline.item_dump()  # Includes version_hash

            # Later, reconstruct with exact versions
            old_filter = factory.get_version(
                "MyFilter",
                config["params"]["filter"]["version_hash"]
            )
        ```

        Note:
            This method does NOT register the fetched version in the
            factory's main registry, avoiding conflicts with the current
            'latest' version. It simply returns the requested class.
        """
        if name in self._foundry and self._version_control.get(name) == code_hash:
            return self._foundry[name]

        rprint(
            f"[cyan]Reconstructing specific version {code_hash[:8]}... for {name}[/cyan]"
        )
        recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)

        return recovered_cls

    def push_all(self) -> None:
        """
        Push all locally registered classes to storage.

        This method iterates through all classes in the factory and
        pushes each one to storage, updating their 'latest' pointers.
        This is the primary way to synchronize local classes with storage.

        Returns:
            None

        Example:
        ```python
            # Register multiple classes
            factory["Filter1"] = Filter1
            factory["Filter2"] = Filter2
            factory["Filter3"] = Filter3

            # Push all at once
            factory.push_all()
        ```

        Note:
            This operation is idempotent - pushing the same class
            multiple times without changes will not create duplicate
            storage entries (same hash = same storage path).
        """
        for name, cls in self._foundry.items():
            rprint(f"[blue]Publishing {name}...[/blue]")
            self._manager.push(cls)

_fallback_factory = fallback_factory instance-attribute

_manager = class_manager instance-attribute

_version_control = {} instance-attribute

__getitem__(name, default=None)

Get class with automatic lazy loading from storage.

This method implements the following fallback chain:

1. Try to get from local memory (parent factory)
2. If not found, check storage for 'latest' version
3. If found in storage, pull and register it
4. If not found anywhere, raise AttributeError

Parameters:

Name Type Description Default
name str

Name of the class to retrieve.

 required

Returns:

Type Description
Type[TypePlugable]

Type[TypePlugable]: The requested class.

Raises:

Type Description
AttributeError

If class not found locally or in storage.

Example: 

    # Class in memory - instant return
    cls = factory["MyFilter"]

    # Class not in memory - auto-pulls from storage
    cls = factory["RemoteFilter"]

Note

Lazy loading is transparent - users don't need to explicitly call pull() before accessing classes.

Source code in labchain/container/persistent/pet_factory.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def __getitem__(
    self, name: str, default: Type[TypePlugable] | None = None
) -> Type[TypePlugable]:
    """
    Get class with automatic lazy loading from storage.

    This method implements the following fallback chain:

        1. Try to get from local memory (parent factory)
        2. If not found, check storage for 'latest' version
        3. If found in storage, pull and register it
        4. If not found anywhere, raise AttributeError

    Args:
        name (str): Name of the class to retrieve.

    Returns:
        Type[TypePlugable]: The requested class.

    Raises:
        AttributeError: If class not found locally or in storage.

    Example:
    ```python
        # Class in memory - instant return
        cls = factory["MyFilter"]

        # Class not in memory - auto-pulls from storage
        cls = factory["RemoteFilter"]
    ```

    Note:
        Lazy loading is transparent - users don't need to explicitly
        call pull() before accessing classes.
    """
    try:
        return super().__getitem__(name)
    except AttributeError:
        # Not in memory, try to fetch 'latest' from storage
        remote_meta = self._manager._get_remote_latest_meta(name)

        if remote_meta:
            code_hash = remote_meta["hash"]
            rprint(
                f"[yellow]Class {name} missing. Pulling latest: {code_hash[:8]}...[/yellow]"
            )

            # Pull and register dynamically
            recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)
            self._version_control[name] = code_hash
            super().__setitem__(name, recovered_cls)
            return recovered_cls

        if name in self._fallback_factory:
            return self._fallback_factory[name]

        raise AttributeError(
            f"Class '{name}' not found locally and no remote version exists."
        )

__init__(class_manager, fallback_factory)

Initialize the PetFactory with a class manager.

Parameters:

Name Type Description Default
class_manager PetClassManager

Manager for storage operations.

 required

Example: 

    manager = PetClassManager(storage)
    factory = PetFactory(manager)

Source code in labchain/container/persistent/pet_factory.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(self, class_manager: PetClassManager, fallback_factory: BaseFactory):
    """
    Initialize the PetFactory with a class manager.

    Args:
        class_manager (PetClassManager): Manager for storage operations.

    Example:
    ```python
        manager = PetClassManager(storage)
        factory = PetFactory(manager)
    ```
    """
    super().__init__()
    os.environ["TYPEGUARD_DISABLE"] = "1"
    self._manager = class_manager
    self._fallback_factory = fallback_factory
    self._version_control: Dict[str, str] = {}

__setitem__(name, value)

Register class locally and compute its content hash.

When a class is registered, this method:

1. Computes the SHA-256 hash of the class source code
2. Stores the hash in version_control for tracking
3. Registers the class in the parent factory

Parameters:

Name Type Description Default
name str

Name to register the class under.

 required
value Type[TypePlugable]

The class to register.

 required

Returns:

Type Description
None

None

Example: 

    factory["MyFilter"] = MyFilter
    # Hash is automatically computed and tracked

Note

The hash is computed immediately but the class is not pushed to storage until push_all() is called.

Source code in labchain/container/persistent/pet_factory.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def __setitem__(self, name: str, value: Type[TypePlugable]) -> None:
    """
    Register class locally and compute its content hash.

    When a class is registered, this method:

        1. Computes the SHA-256 hash of the class source code
        2. Stores the hash in version_control for tracking
        3. Registers the class in the parent factory

    Args:
        name (str): Name to register the class under.
        value (Type[TypePlugable]): The class to register.

    Returns:
        None

    Example:
    ```python
        factory["MyFilter"] = MyFilter
        # Hash is automatically computed and tracked
    ```

    Note:
        The hash is computed immediately but the class is not pushed
        to storage until push_all() is called.
    """
    # code_hash = self._manager.get_class_hash(value)
    class_hash = cast(BasePlugin, value)._hash
    if class_hash is not None:
        self._version_control[name] = class_hash
        super().__setitem__(name, value)
    else:
        raise ValueError("Class must be a BasePlugin and should have a hash")

get(name, default=None)

Get class with optional default, attempting lazy load from storage.

This method overrides the parent's get() to add lazy loading capability. It provides a safe way to retrieve classes with a fallback value.

Parameters:

Name Type Description Default
name str

Name of the class to retrieve.

 required
default Optional[Type[TypePlugable]]

Default value if not found.

 None

Returns:

Type Description
Type[TypePlugable] | None

Optional[Type[TypePlugable]]: The class or default value.

Example: 

    # With default
    cls = factory.get("MaybeExists", DefaultFilter)

    # Without default (returns None if not found)
    cls = factory.get("MaybeExists")
    if cls is None:
        print("Class not found")

Note

Unlike getitem, this method returns the default value (or None) instead of raising an exception when the class is not found.

Source code in labchain/container/persistent/pet_factory.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def get(
    self, name: str, default: Type[TypePlugable] | None = None
) -> Type[TypePlugable] | None:
    """
    Get class with optional default, attempting lazy load from storage.

    This method overrides the parent's get() to add lazy loading capability.
    It provides a safe way to retrieve classes with a fallback value.

    Args:
        name (str): Name of the class to retrieve.
        default (Optional[Type[TypePlugable]]): Default value if not found.

    Returns:
        Optional[Type[TypePlugable]]: The class or default value.

    Example:
    ```python
        # With default
        cls = factory.get("MaybeExists", DefaultFilter)

        # Without default (returns None if not found)
        cls = factory.get("MaybeExists")
        if cls is None:
            print("Class not found")
    ```

    Note:
        Unlike __getitem__, this method returns the default value
        (or None) instead of raising an exception when the class
        is not found.
    """
    try:
        return self.__getitem__(name)
    except AttributeError:
        return default

get_version(name, code_hash)

Get a specific version of a class by its hash.

This method is essential for reconstructing old pipelines via BasePlugin.build_from_dump(). It allows loading exact versions of classes as they existed when a pipeline was saved.

Parameters:

Name Type Description Default
name str

Name of the class.

 required
code_hash str

SHA-256 hash of the specific version.

 required

Returns:

Type Description
Type[TypePlugable]

Type[TypePlugable]: The requested version of the class.

Example: 

    # Save pipeline with current versions
    config = pipeline.item_dump()  # Includes version_hash

    # Later, reconstruct with exact versions
    old_filter = factory.get_version(
        "MyFilter",
        config["params"]["filter"]["version_hash"]
    )

Note

This method does NOT register the fetched version in the factory's main registry, avoiding conflicts with the current 'latest' version. It simply returns the requested class.

Source code in labchain/container/persistent/pet_factory.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def get_version(self, name: str, code_hash: str) -> Type[TypePlugable]:
    """
    Get a specific version of a class by its hash.

    This method is essential for reconstructing old pipelines via
    BasePlugin.build_from_dump(). It allows loading exact versions
    of classes as they existed when a pipeline was saved.

    Args:
        name (str): Name of the class.
        code_hash (str): SHA-256 hash of the specific version.

    Returns:
        Type[TypePlugable]: The requested version of the class.

    Example:
    ```python
        # Save pipeline with current versions
        config = pipeline.item_dump()  # Includes version_hash

        # Later, reconstruct with exact versions
        old_filter = factory.get_version(
            "MyFilter",
            config["params"]["filter"]["version_hash"]
        )
    ```

    Note:
        This method does NOT register the fetched version in the
        factory's main registry, avoiding conflicts with the current
        'latest' version. It simply returns the requested class.
    """
    if name in self._foundry and self._version_control.get(name) == code_hash:
        return self._foundry[name]

    rprint(
        f"[cyan]Reconstructing specific version {code_hash[:8]}... for {name}[/cyan]"
    )
    recovered_cls: Type[TypePlugable] = self._manager.pull(name, code_hash)

    return recovered_cls

push_all()

Push all locally registered classes to storage.

This method iterates through all classes in the factory and pushes each one to storage, updating their 'latest' pointers. This is the primary way to synchronize local classes with storage.

Returns:

Type Description
None

None

Example: 

    # Register multiple classes
    factory["Filter1"] = Filter1
    factory["Filter2"] = Filter2
    factory["Filter3"] = Filter3

    # Push all at once
    factory.push_all()

Note

This operation is idempotent - pushing the same class multiple times without changes will not create duplicate storage entries (same hash = same storage path).

Source code in labchain/container/persistent/pet_factory.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def push_all(self) -> None:
    """
    Push all locally registered classes to storage.

    This method iterates through all classes in the factory and
    pushes each one to storage, updating their 'latest' pointers.
    This is the primary way to synchronize local classes with storage.

    Returns:
        None

    Example:
    ```python
        # Register multiple classes
        factory["Filter1"] = Filter1
        factory["Filter2"] = Filter2
        factory["Filter3"] = Filter3

        # Push all at once
        factory.push_all()
    ```

    Note:
        This operation is idempotent - pushing the same class
        multiple times without changes will not create duplicate
        storage entries (same hash = same storage path).
    """
    for name, cls in self._foundry.items():
        rprint(f"[blue]Publishing {name}...[/blue]")
        self._manager.push(cls)