Skip to content

graflo.data_source.base

Base classes for data source abstraction.

This module defines the abstract base class and types for all data sources. Data sources handle data retrieval from various sources (files, APIs, databases) and provide a unified interface for batch iteration.

Classes

AbstractDataSource

Bases: ConfigBaseModel, ABC

Abstract base class for all data sources.

Data sources handle data retrieval from various sources and provide a unified interface for batch iteration. They are separate from Resources, which handle data transformation. Many DataSources can map to the same Resource.

Attributes:

Name Type Description
source_type DataSourceType

Type of the data source
resource_name str | None

Name of the resource this data source maps to (set externally via DataSourceRegistry)
Source code in graflo/data_source/base.py 
 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
class AbstractDataSource(ConfigBaseModel, abc.ABC):
    """Abstract base class for all data sources.

    Data sources handle data retrieval from various sources and provide
    a unified interface for batch iteration. They are separate from Resources,
    which handle data transformation. Many DataSources can map to the same Resource.

    Attributes:
        source_type: Type of the data source
        resource_name: Name of the resource this data source maps to
            (set externally via DataSourceRegistry)
    """

    source_type: DataSourceType
    _resource_name: str | None = PrivateAttr(default=None)

    @property
    def resource_name(self) -> str | None:
        """Get the resource name this data source maps to.

        Returns:
            Resource name or None if not set
        """
        return self._resource_name

    @resource_name.setter
    def resource_name(self, value: str | None) -> None:
        """Set the resource name this data source maps to.

        Args:
            value: Resource name to set
        """
        self._resource_name = value

    @property
    def high_water(self) -> Any:
        """Largest watermark value the current iteration has yielded.

        ``None`` for a source without a declared watermark.
        """
        return None

    @abc.abstractmethod
    def iter_batches(
        self, batch_size: int = 1000, limit: int | None = None
    ) -> Iterator[list[dict]]:
        """Iterate over data in batches.

        This method yields batches of documents (dictionaries) from the data source.
        Each batch is a list of dictionaries representing the data items.

        Args:
            batch_size: Maximum number of items per yielded batch (last batch may
                be smaller).
            limit: Cap on **total items** read from this source (rows, JSON
                documents, SPARQL subjects after grouping, etc.), not a cap on
                batches. ``None`` means read until the source is exhausted.

        Yields:
            list[dict]: Batches of documents as dictionaries

        Raises:
            NotImplementedError: Must be implemented by subclasses
        """
        raise NotImplementedError("Subclasses must implement iter_batches")

    def acknowledge(self, batch_index: int) -> None:
        """Say that a batch has been written to the target.

        A source that keeps a read position outside the process (a Kafka
        consumer group) advances it only past acknowledged batches, so a batch
        that was read and never written is read again. Other sources ignore
        the call.

        Args:
            batch_index: Position of the batch among those the current
                :meth:`iter_batches` call yielded, from 0.
        """

    def close(self) -> None:
        """Release what :meth:`iter_batches` opened. Safe to call more than once."""

    def __iter__(self):
        """Make data source iterable, yielding individual items.

        Yields:
            dict: Individual documents
        """
        for batch in self.iter_batches(batch_size=1, limit=None):
            yield from batch

Attributes

high_water property

Largest watermark value the current iteration has yielded.

None for a source without a declared watermark.

resource_name property writable

Get the resource name this data source maps to.

Returns:

Type Description
str | None

Resource name or None if not set
source_type instance-attribute

Methods:

__iter__()

Make data source iterable, yielding individual items.

Yields:

Name Type Description
dict

Individual documents
Source code in graflo/data_source/base.py 
143
144
145
146
147
148
149
150
def __iter__(self):
    """Make data source iterable, yielding individual items.

    Yields:
        dict: Individual documents
    """
    for batch in self.iter_batches(batch_size=1, limit=None):
        yield from batch
acknowledge(batch_index)

Say that a batch has been written to the target.

A source that keeps a read position outside the process (a Kafka consumer group) advances it only past acknowledged batches, so a batch that was read and never written is read again. Other sources ignore the call.

Parameters:

Name Type Description Default
batch_index int

Position of the batch among those the current :meth:iter_batches call yielded, from 0.

 required
Source code in graflo/data_source/base.py 
127
128
129
130
131
132
133
134
135
136
137
138
def acknowledge(self, batch_index: int) -> None:
    """Say that a batch has been written to the target.

    A source that keeps a read position outside the process (a Kafka
    consumer group) advances it only past acknowledged batches, so a batch
    that was read and never written is read again. Other sources ignore
    the call.

    Args:
        batch_index: Position of the batch among those the current
            :meth:`iter_batches` call yielded, from 0.
    """
close()

Release what :meth:iter_batches opened. Safe to call more than once.

Source code in graflo/data_source/base.py 
140
141
def close(self) -> None:
    """Release what :meth:`iter_batches` opened. Safe to call more than once."""
iter_batches(batch_size=1000, limit=None) abstractmethod

Iterate over data in batches.

This method yields batches of documents (dictionaries) from the data source. Each batch is a list of dictionaries representing the data items.

Parameters:

Name Type Description Default
batch_size int

Maximum number of items per yielded batch (last batch may be smaller).

 1000
limit int | None

Cap on total items read from this source (rows, JSON documents, SPARQL subjects after grouping, etc.), not a cap on batches. None means read until the source is exhausted.

 None

Yields:

Type Description
list[dict]

list[dict]: Batches of documents as dictionaries

Raises:

Type Description
NotImplementedError

Must be implemented by subclasses
Source code in graflo/data_source/base.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@abc.abstractmethod
def iter_batches(
    self, batch_size: int = 1000, limit: int | None = None
) -> Iterator[list[dict]]:
    """Iterate over data in batches.

    This method yields batches of documents (dictionaries) from the data source.
    Each batch is a list of dictionaries representing the data items.

    Args:
        batch_size: Maximum number of items per yielded batch (last batch may
            be smaller).
        limit: Cap on **total items** read from this source (rows, JSON
            documents, SPARQL subjects after grouping, etc.), not a cap on
            batches. ``None`` means read until the source is exhausted.

    Yields:
        list[dict]: Batches of documents as dictionaries

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError("Subclasses must implement iter_batches")

DataSourceType

Bases: BaseEnum

Types of data sources supported by the system.

FILE: File-based data sources (JSON, JSONL, CSV/TSV) API: REST API data sources SQL: SQL database data sources IN_MEMORY: In-memory data sources (lists, DataFrames) SPARQL: RDF data sources (local files via rdflib, remote endpoints via SPARQLWrapper) KAFKA: Kafka topic consumers (JSON object payloads)

Source code in graflo/data_source/base.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class DataSourceType(BaseEnum):
    """Types of data sources supported by the system.

    FILE: File-based data sources (JSON, JSONL, CSV/TSV)
    API: REST API data sources
    SQL: SQL database data sources
    IN_MEMORY: In-memory data sources (lists, DataFrames)
    SPARQL: RDF data sources (local files via rdflib, remote endpoints via SPARQLWrapper)
    KAFKA: Kafka topic consumers (JSON object payloads)
    """

    FILE = "file"
    API = "api"
    SQL = "sql"
    IN_MEMORY = "in_memory"
    SPARQL = "sparql"
    KAFKA = "kafka"

Attributes

API = 'api' class-attribute instance-attribute
FILE = 'file' class-attribute instance-attribute
IN_MEMORY = 'in_memory' class-attribute instance-attribute
KAFKA = 'kafka' class-attribute instance-attribute
SPARQL = 'sparql' class-attribute instance-attribute
SQL = 'sql' class-attribute instance-attribute

Functions:

later_watermark(candidate, current)

True when candidate is past current as a watermark.

Two numbers (or numeric strings) compare as numbers, anything else as text.

Source code in graflo/data_source/base.py 
50
51
52
53
54
55
56
57
58
def later_watermark(candidate: object, current: object) -> bool:
    """True when *candidate* is past *current* as a watermark.

    Two numbers (or numeric strings) compare as numbers, anything else as text.
    """
    a, b = _as_number(candidate), _as_number(current)
    if a is not None and b is not None:
        return a > b
    return str(candidate) > str(current)