Class DataSource

python/pathway/internals/datasource.py:32–69

@dataclass(frozen=True, kw_only=True)
class DataSource(ABC):
    schema: type[Schema]
    data_source_options: DataSourceOptions = DataSourceOptions()

    @property
    def connector_properties(self) -> api.ConnectorProperties:
        columns: list[api.ColumnProperties] = []
        for column in self.schema.columns().values():
            columns.append(
                api.ColumnProperties(
                    dtype=column.dtype.to_engine(),
                    append_only=self.is_append_only(),
                )
            )

        return api.ConnectorProperties(
            commit_duration_ms=self.data_source_options.commit_duration_ms,
            unsafe_trusted_ids=self.data_source_options.unsafe_trusted_ids,
            column_properties=columns,
            unique_name=self.data_source_options.unique_name,
            synchronization_group=self.data_source_options.synchronization_group,
            max_backlog_size=self.data_source_options.max_backlog_size,
        )

    def get_effective_schema(self) -> type[Schema]:
        if self.is_append_only():
            return self.schema.update_properties(append_only=True)
        return self.schema

    @abstractmethod
    def is_bounded(self) -> bool: ...

    @abstractmethod
    def is_append_only(self) -> bool: ...

    @property
    def name(self) -> str:
        return type(self).__qualname__.lower().removesuffix("datasource")


class StaticDataSource(DataSource, ABC):