Skip to content

base_loader

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

F = TypeVar('F') module-attribute

S = TypeVar('S') module-attribute

BaseLoader

Bases: FuturesExecutorMixin, ABC, Generic[S, F, A]

Fetches data from a source, parses it, and returns an Artifact.

Attributes:

Name Type Description
reference Optional[Reference]

The optional Reference to set on the Artifact.
Source code in griptape/loaders/base_loader.py 
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
@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)

        artifact = self.parse(data)

        artifact.reference = self.reference

        return artifact

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

    ...

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

    ...

    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}

        return execute_futures_dict(
            {key: self.futures_executor.submit(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)
        else:
            return str_to_hash(str(source))

reference: Optional[Reference] = field(default=None, kw_only=True) class-attribute instance-attribute

fetch(source) abstractmethod

Fetches data from the source.

Source code in griptape/loaders/base_loader.py 
42
43
44
@abstractmethod
def fetch(self, source: S) -> F:
    """Fetches data from the source."""

load(source)

Source code in griptape/loaders/base_loader.py 
33
34
35
36
37
38
39
40
def load(self, source: S) -> A:
    data = self.fetch(source)

    artifact = self.parse(data)

    artifact.reference = self.reference

    return artifact

load_collection(sources)

Loads a collection of sources and returns a dictionary of Artifacts.

Source code in griptape/loaders/base_loader.py 
54
55
56
57
58
59
60
61
62
63
64
65
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}

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

parse(data) abstractmethod

Parses the fetched data and returns an Artifact.

Source code in griptape/loaders/base_loader.py 
48
49
50
@abstractmethod
def parse(self, data: F) -> A:
    """Parses the fetched data and returns an Artifact."""

to_key(source)

Converts the source to a key for the collection.

Source code in griptape/loaders/base_loader.py 
67
68
69
70
71
72
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)
    else:
        return str_to_hash(str(source))