base_loader

base_loader#

  • A = TypeVar('A', bound=BaseArtifact) module-attribute

  • F = TypeVar('F') module-attribute

  • S = TypeVar('S') module-attribute

Bases: FuturesExecutorMixin “) , ABC, Generic[S, F, A

Attributes#

Name

Type

Description

<a href="base_loader.mdx#griptape.loaders.base_loader.BaseLoader.reference" title="reference = field(default=None, kw_only=True) class-attribute instance-attribute (griptape.loaders.base_loader.BaseLoader.reference)"><code class="language-python">reference</code></a>

<a href="../common#griptape.common.Reference" title="Reference (griptape.common.Reference)"><code class="language-python">Optional[Reference]</code></a>

The optional Reference to set on the Artifact.<br/>

Source Code in griptape/loaders/base_loader.py

@define
class BaseLoader(FuturesExecutorMixin, ABC, Generic[S, F, A]):
    """Fetches data from a source, parses it, and returns an Artifact.

    Attributes:
        reference: The optional `Reference` to set on the Artifact.
    """

    reference: Optional[Reference] = field(default=None, kw_only=True)

    def load(self, source: S) -> A:
        data = self.fetch(source)

        return self.parse(data)

    @abstractmethod
    def fetch(self, source: S) -> F:
        """Fetches data from the source."""

    def parse(self, data: F) -> A:
        """Parses the fetched data and returns an Artifact."""
        artifact = self.try_parse(data)

        artifact.reference = self.reference

        return artifact

    def try_parse(self, data: F) -> A:
        """Parses the fetched data and returns an Artifact."""
        # TODO: Mark as abstract method for griptape 2.0
        raise NotImplementedError

    def load_collection(
        self,
        sources: list[Any],
    ) -> Mapping[str, A]:
        """Loads a collection of sources and returns a dictionary of Artifacts."""
        # Create a dictionary before actually submitting the jobs to the executor
        # to avoid duplicate work.
        sources_by_key = {self.to_key(source): source for source in sources}

        with self.create_futures_executor() as futures_executor:
            return execute_futures_dict(
                {
                    key: futures_executor.submit(with_contextvars(self.load), source)
                    for key, source in sources_by_key.items()
                },
            )

    def to_key(self, source: S) -> str:
        """Converts the source to a key for the collection."""
        if isinstance(source, bytes):
            return bytes_to_hash(source)
        return str_to_hash(str(source))

  • reference = field(default=None, kw_only=True) class-attribute instance-attribute

fetch(source)abstractmethod#

Source Code in griptape/loaders/base_loader.py

@abstractmethod
def fetch(self, source: S) -> F:
    """Fetches data from the source."""

load(source)#

Source Code in griptape/loaders/base_loader.py

def load(self, source: S) -> A:
    data = self.fetch(source)

    return self.parse(data)

load_collection(sources)#

Source Code in griptape/loaders/base_loader.py

def load_collection(
    self,
    sources: list[Any],
) -> Mapping[str, A]:
    """Loads a collection of sources and returns a dictionary of Artifacts."""
    # Create a dictionary before actually submitting the jobs to the executor
    # to avoid duplicate work.
    sources_by_key = {self.to_key(source): source for source in sources}

    with self.create_futures_executor() as futures_executor:
        return execute_futures_dict(
            {
                key: futures_executor.submit(with_contextvars(self.load), source)
                for key, source in sources_by_key.items()
            },
        )

parse(data)#

Source Code in griptape/loaders/base_loader.py

def parse(self, data: F) -> A:
    """Parses the fetched data and returns an Artifact."""
    artifact = self.try_parse(data)

    artifact.reference = self.reference

    return artifact

to_key(source)#

Source Code in griptape/loaders/base_loader.py

def to_key(self, source: S) -> str:
    """Converts the source to a key for the collection."""
    if isinstance(source, bytes):
        return bytes_to_hash(source)
    return str_to_hash(str(source))

try_parse(data)#

Source Code in griptape/loaders/base_loader.py

def try_parse(self, data: F) -> A:
    """Parses the fetched data and returns an Artifact."""
    # TODO: Mark as abstract method for griptape 2.0
    raise NotImplementedError
Contents