Storage

Storage module.

Modules:

Name	Description
local	Local storage client.

Storage

Bases: Protocol

Storage client protocol.

Source code in devpipe/storage/__init__.py

class Storage(Protocol):
    """Storage client protocol."""

    def read(self, storage_key: str) -> bytes:
        """Read data from storage.

        Args:
            storage_key (str): Storage key.

        Returns:
            Stored data.
        """

    def save(self, storage_key: str, data: bytes) -> None:
        """Save data to storage.

        Args:
            storage_key (str): Storage key.
            data (bytes): Data to store.
        """

read(storage_key)

Read data from storage.

Parameters:

Name	Type	Description	Default
storage_key	str	Storage key.	required

Returns:

Type	Description
bytes	Stored data.

Source code in devpipe/storage/__init__.py

def read(self, storage_key: str) -> bytes:
    """Read data from storage.

    Args:
        storage_key (str): Storage key.

    Returns:
        Stored data.
    """

save(storage_key, data)

Save data to storage.

Parameters:

Name	Type	Description	Default
storage_key	str	Storage key.	required
data	bytes	Data to store.	required

Source code in devpipe/storage/__init__.py

def save(self, storage_key: str, data: bytes) -> None:
    """Save data to storage.

    Args:
        storage_key (str): Storage key.
        data (bytes): Data to store.
    """

set_storage(storage=None)

Set storage client.

Parameters:

Name	Type	Description	Default
storage	Optional[Storage]	Storage client. If not provided, a local storage client will be used.	None

Example

import devpipe as dp
from devpipe.storage import set_storage
from devpipe.storage.local import LocalStorage

storage = LocalStorage()
set_storage(storage)

@dp.pipeline
def my_pipeline():
    ...

Source code in devpipe/storage/__init__.py

def set_storage(storage: Optional[Storage] = None) -> Storage:
    """Set storage client.

    Args:
        storage (Optional[Storage], optional): Storage client. If not
            provided, a local storage client will be used.

    Example:
        ```python
        import devpipe as dp
        from devpipe.storage import set_storage
        from devpipe.storage.local import LocalStorage

        storage = LocalStorage()
        set_storage(storage)

        @dp.pipeline
        def my_pipeline():
            ...
        ```
    """
    if storage is None:
        storage = local.LocalStorage()
    DEVPIPE["storage"] = storage
    logger.debug(f"Storage set to `{storage.__class__.__name__}`.")
    return DEVPIPE["storage"]