-
Notifications
You must be signed in to change notification settings - Fork 4
/
s3_sink.py
28 lines (26 loc) · 981 Bytes
/
s3_sink.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from streams_explorer.core.extractor.extractor import ConnectorExtractor
from streams_explorer.models.kafka_connector import (
KafkaConnector,
KafkaConnectorTypesEnum,
)
from streams_explorer.models.sink import Sink
class S3Sink(ConnectorExtractor):
def on_connector_info_parsing(
self, info: dict, connector_name: str
) -> KafkaConnector | None:
config = info["config"]
connector_class = config.get("connector.class")
if connector_class and "S3SinkConnector" in connector_class:
if name := config.get("s3.bucket.name"):
self.sinks.append(
Sink(
name=name,
node_type="s3-bucket",
source=connector_name,
)
)
return KafkaConnector(
name=connector_name,
config=config,
type=KafkaConnectorTypesEnum.SINK,
)