sift_py.ingestion.manager

CLASS	DESCRIPTION
IngestionConfigServiceBuilderIdentifierMap
IngestionServiceManagerError
IngestionServicesManager	Allows for the initialization of multiple instances of sift_py.ingestion.service.IngestionService from
TelemetryConfigByIdentifierMap

ATTRIBUTE	DESCRIPTION
IngestionServiceBuilder	TYPE: TypeAlias

IngestionServiceBuilder module-attribute

IngestionServiceBuilder: TypeAlias = Callable[
    [SiftChannel], IngestionService
]

IngestionConfigServiceBuilderIdentifierMap

Bases: TypedDict

ATTRIBUTE	DESCRIPTION
builder	TYPE: IngestionServiceBuilder
identifier	TYPE: str

builder instance-attribute

builder: IngestionServiceBuilder

identifier instance-attribute

identifier: str

IngestionServiceManagerError

IngestionServiceManagerError(msg: str)

Bases: Exception

IngestionServicesManager

IngestionServicesManager(
    transport_channel: SiftChannel,
    ingestion_services: Dict[str, IngestionService],
)

Allows for the initialization of multiple instances of sift_py.ingestion.service.IngestionService from either telemetry configs or builders under a single wrapper class that assists in managing data-ingestion for multiple telemetry configs.

The initializer of this class can be used directly, but prefer to use either from_builders or from_telemetry_configs. Prefer to use from_builders if you have custom options that you want to provide to sift_py.ingestion.service.IngestionService.__init__.

Example usage:

manager = IngestionServicesManager.from_telementry_configs(grpc_channel, {
    "config_a": config_a,
    "config_b": config_b,
})

with manager.ingestion_service("config_a") as config_a:
    config_a.try_ingest_flow(...)

with manager.ingestion_service("config_b") as config_b:
    config_b.try_ingest_flow(...)

METHOD	DESCRIPTION
__getitem__
from_builders
from_telemetry_configs
get_ingestion_service_by_identifier
ingestion_service

__getitem__

__getitem__(identifier: str) -> Optional[IngestionService]

from_builders classmethod

from_builders(
    channel: SiftChannel,
    builders: Dict[str, IngestionServiceBuilder],
) -> Self

from_telemetry_configs classmethod

from_telemetry_configs(
    channel: SiftChannel,
    telemetry_configs: Dict[str, TelemetryConfig],
) -> Self

get_ingestion_service_by_identifier

get_ingestion_service_by_identifier(
    identifier: str,
) -> Optional[IngestionService]

ingestion_service

ingestion_service(
    identifier: str,
) -> Iterator[IngestionService]

TelemetryConfigByIdentifierMap

Bases: TypedDict

ATTRIBUTE	DESCRIPTION
identifier	TYPE: str
telemetry_config	TYPE: TelemetryConfig

identifier instance-attribute

identifier: str

telemetry_config instance-attribute

telemetry_config: TelemetryConfig