forked from feldera/feldera
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconnector.py
More file actions
33 lines (26 loc) · 1.02 KB
/
Copy pathconnector.py
File metadata and controls
33 lines (26 loc) · 1.02 KB
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
29
30
31
32
33
from typing import Any, Mapping, Optional
from feldera.rest.attached_connector import AttachedConnector
class Connector:
"""
A generic connector class that can be used to represent any Feldera connector
"""
def __init__(
self,
name: str,
description: Optional[str] = None,
config: Optional[Mapping[str, Any]] = None,
id: Optional[str] = None,
):
"""
Create a new connector
:param name: The name of the connector
:param description: A description of the connector
:param config: The configuration of the connector
:param id: Optional. The ID of the connector. Not to be set by the user
"""
self.name: str = name
self.config: Mapping[str, Any] = config or {}
self.description: Optional[str] = description
self.id: Optional[str] = id
def attach_relation(self, relation_name: str, is_input: bool) -> AttachedConnector:
return AttachedConnector(self.name, relation_name, is_input)