Function _create_python_datasource

python/pathway/io/python/__init__.py:331–373

(
    subject: ConnectorSubject,
    *,
    schema: type[Schema],
    autocommit_duration_ms: int | None = 1500,
    name: str | None = None,
    max_backlog_size: int | None = None,
    _stacklevel: int = 1,
    **kwargs,
)

def _create_python_datasource(
    subject: ConnectorSubject,
    *,
    schema: type[Schema],
    autocommit_duration_ms: int | None = 1500,
    name: str | None = None,
    max_backlog_size: int | None = None,
    _stacklevel: int = 1,
    **kwargs,
) -> datasource.GenericDataSource:
    schema, api_schema = read_schema(schema)
    data_format = api.DataFormat(
        **api_schema,
        format_type="transparent",
        session_type=subject._session_type,
    )
    if max_backlog_size is not None:
        subject._set_max_backlog_size(max_backlog_size)
    data_storage = api.DataStorage(
        storage_type="python",
        python_subject=api.PythonSubject(
            start=subject.start,
            seek=subject.seek,
            on_persisted_run=subject.on_persisted_run,
            read=subject._read,
            end=subject.end,
            is_internal=subject._is_internal(),
            deletions_enabled=subject._deletions_enabled,
        ),
    )
    data_source_options = datasource.DataSourceOptions(
        commit_duration_ms=autocommit_duration_ms,
        max_backlog_size=max_backlog_size,
        unique_name=_get_unique_name(name, kwargs, stacklevel=_stacklevel + 1),
    )
    return datasource.GenericDataSource(
        datastorage=data_storage,
        dataformat=data_format,
        data_source_options=data_source_options,
        schema=schema,
        datasource_name=subject._datasource_name,
        append_only=not subject._deletions_enabled,
    )


def _are_deletions_reachable(subject) -> bool: