Skip to content

graflo.connections.sources

Pydantic models for runtime source connection configuration.

Leaf module: no imports from bindings, data_source, hq, or :mod:graflo.connections.provider. Used by :mod:graflo.connections.provider, :mod:graflo.data_source.api, and :mod:graflo.data_source.kafka.

Attributes

AuthType = Literal['bearer', 'basic', 'digest', 'api_key'] module-attribute

GeneralizedConnConfig = PostgresGeneralizedConnConfig | SparqlGeneralizedConnConfig | S3GeneralizedConnConfig | ApiGeneralizedConnConfig | KafkaGeneralizedConnConfig module-attribute

KafkaSecurityProtocol = Literal['PLAINTEXT', 'SASL_PLAINTEXT', 'SASL_SSL', 'SSL'] module-attribute

Classes

ApiAuth

Bases: BaseModel

Authentication payload for REST API source access.

Source code in graflo/connections/sources.py 
28
29
30
31
32
33
34
35
36
class ApiAuth(BaseModel):
    """Authentication payload for REST API source access."""

    auth_type: Literal["bearer", "basic", "digest", "api_key"] = "bearer"
    token: str | None = None
    username: str | None = None
    password: str | None = None
    header_name: str = "Authorization"
    prefix: str = "Bearer"

Attributes

auth_type = 'bearer' class-attribute instance-attribute
header_name = 'Authorization' class-attribute instance-attribute
password = None class-attribute instance-attribute
prefix = 'Bearer' class-attribute instance-attribute
token = None class-attribute instance-attribute
username = None class-attribute instance-attribute

ApiGeneralizedConnConfig

Bases: BaseModel

Generalized runtime config variant for REST API connections.

Source code in graflo/connections/sources.py 
135
136
137
138
139
class ApiGeneralizedConnConfig(BaseModel):
    """Generalized runtime config variant for REST API connections."""

    kind: Literal["rest_api"] = "rest_api"
    config: RestApiConnConfig

Attributes

config instance-attribute
kind = 'rest_api' class-attribute instance-attribute

KafkaConnConfig

Bases: BaseModel

Runtime Kafka broker connection settings (bootstrap and auth).

Source code in graflo/connections/sources.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class KafkaConnConfig(BaseModel):
    """Runtime Kafka broker connection settings (bootstrap and auth)."""

    bootstrap_servers: str
    security_protocol: KafkaSecurityProtocol = "PLAINTEXT"
    client_id: str | None = None
    sasl_mechanism: str | None = None
    sasl_username: str | None = None
    sasl_password: str | None = None

    @classmethod
    def from_env(cls, env_prefix: str) -> KafkaConnConfig:
        """Load Kafka config from environment variables.

        Supported variables (all prefixed with *env_prefix*):

        - ``BOOTSTRAP_SERVERS`` (required)
        - ``SECURITY_PROTOCOL``: ``PLAINTEXT``, ``SASL_PLAINTEXT``, ``SASL_SSL``,
          or ``SSL`` (default: ``PLAINTEXT``)
        - ``CLIENT_ID``
        - ``SASL_MECHANISM``, ``SASL_USERNAME``, ``SASL_PASSWORD``
        """
        bootstrap_servers = os.environ.get(f"{env_prefix}BOOTSTRAP_SERVERS")
        if not bootstrap_servers:
            raise ValueError(
                f"Environment variable {env_prefix}BOOTSTRAP_SERVERS is required "
                "for KafkaConnConfig"
            )

        security_raw = os.environ.get(f"{env_prefix}SECURITY_PROTOCOL", "PLAINTEXT")
        security_upper = security_raw.upper()
        if security_upper not in _VALID_KAFKA_SECURITY_PROTOCOLS:
            raise ValueError(
                f"Invalid {env_prefix}SECURITY_PROTOCOL={security_raw!r}; "
                "expected PLAINTEXT, SASL_PLAINTEXT, SASL_SSL, or SSL"
            )

        return cls(
            bootstrap_servers=cast(str, bootstrap_servers),
            security_protocol=cast(KafkaSecurityProtocol, security_upper),
            client_id=os.environ.get(f"{env_prefix}CLIENT_ID"),
            sasl_mechanism=os.environ.get(f"{env_prefix}SASL_MECHANISM"),
            sasl_username=os.environ.get(f"{env_prefix}SASL_USERNAME"),
            sasl_password=os.environ.get(f"{env_prefix}SASL_PASSWORD"),
        )

    def to_consumer_config(self) -> dict[str, str]:
        """Build a confluent-kafka consumer config dict from connection settings."""
        cfg: dict[str, str] = {
            "bootstrap.servers": self.bootstrap_servers,
            "security.protocol": self.security_protocol,
        }
        if self.client_id:
            cfg["client.id"] = self.client_id
        if self.sasl_mechanism:
            cfg["sasl.mechanism"] = self.sasl_mechanism
        if self.sasl_username is not None:
            cfg["sasl.username"] = self.sasl_username
        if self.sasl_password is not None:
            cfg["sasl.password"] = self.sasl_password
        return cfg

Attributes

bootstrap_servers instance-attribute
client_id = None class-attribute instance-attribute
sasl_mechanism = None class-attribute instance-attribute
sasl_password = None class-attribute instance-attribute
sasl_username = None class-attribute instance-attribute
security_protocol = 'PLAINTEXT' class-attribute instance-attribute

Methods:

from_env(env_prefix) classmethod

Load Kafka config from environment variables.

Supported variables (all prefixed with env_prefix):

  • BOOTSTRAP_SERVERS (required)
  • SECURITY_PROTOCOL: PLAINTEXT, SASL_PLAINTEXT, SASL_SSL, or SSL (default: PLAINTEXT)
  • CLIENT_ID
  • SASL_MECHANISM, SASL_USERNAME, SASL_PASSWORD
Source code in graflo/connections/sources.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@classmethod
def from_env(cls, env_prefix: str) -> KafkaConnConfig:
    """Load Kafka config from environment variables.

    Supported variables (all prefixed with *env_prefix*):

    - ``BOOTSTRAP_SERVERS`` (required)
    - ``SECURITY_PROTOCOL``: ``PLAINTEXT``, ``SASL_PLAINTEXT``, ``SASL_SSL``,
      or ``SSL`` (default: ``PLAINTEXT``)
    - ``CLIENT_ID``
    - ``SASL_MECHANISM``, ``SASL_USERNAME``, ``SASL_PASSWORD``
    """
    bootstrap_servers = os.environ.get(f"{env_prefix}BOOTSTRAP_SERVERS")
    if not bootstrap_servers:
        raise ValueError(
            f"Environment variable {env_prefix}BOOTSTRAP_SERVERS is required "
            "for KafkaConnConfig"
        )

    security_raw = os.environ.get(f"{env_prefix}SECURITY_PROTOCOL", "PLAINTEXT")
    security_upper = security_raw.upper()
    if security_upper not in _VALID_KAFKA_SECURITY_PROTOCOLS:
        raise ValueError(
            f"Invalid {env_prefix}SECURITY_PROTOCOL={security_raw!r}; "
            "expected PLAINTEXT, SASL_PLAINTEXT, SASL_SSL, or SSL"
        )

    return cls(
        bootstrap_servers=cast(str, bootstrap_servers),
        security_protocol=cast(KafkaSecurityProtocol, security_upper),
        client_id=os.environ.get(f"{env_prefix}CLIENT_ID"),
        sasl_mechanism=os.environ.get(f"{env_prefix}SASL_MECHANISM"),
        sasl_username=os.environ.get(f"{env_prefix}SASL_USERNAME"),
        sasl_password=os.environ.get(f"{env_prefix}SASL_PASSWORD"),
    )
to_consumer_config()

Build a confluent-kafka consumer config dict from connection settings.

Source code in graflo/connections/sources.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def to_consumer_config(self) -> dict[str, str]:
    """Build a confluent-kafka consumer config dict from connection settings."""
    cfg: dict[str, str] = {
        "bootstrap.servers": self.bootstrap_servers,
        "security.protocol": self.security_protocol,
    }
    if self.client_id:
        cfg["client.id"] = self.client_id
    if self.sasl_mechanism:
        cfg["sasl.mechanism"] = self.sasl_mechanism
    if self.sasl_username is not None:
        cfg["sasl.username"] = self.sasl_username
    if self.sasl_password is not None:
        cfg["sasl.password"] = self.sasl_password
    return cfg

KafkaGeneralizedConnConfig

Bases: BaseModel

Generalized runtime config variant for Kafka connections.

Source code in graflo/connections/sources.py 
211
212
213
214
215
class KafkaGeneralizedConnConfig(BaseModel):
    """Generalized runtime config variant for Kafka connections."""

    kind: Literal["kafka"] = "kafka"
    config: KafkaConnConfig

Attributes

config instance-attribute
kind = 'kafka' class-attribute instance-attribute

PostgresGeneralizedConnConfig

Bases: BaseModel

Generalized runtime config variant for SQL/Postgres connections.

Source code in graflo/connections/sources.py 
 97
 98
 99
100
101
class PostgresGeneralizedConnConfig(BaseModel):
    """Generalized runtime config variant for SQL/Postgres connections."""

    kind: Literal["postgres"] = "postgres"
    config: PostgresConfig

Attributes

config instance-attribute
kind = 'postgres' class-attribute instance-attribute

RestApiConnConfig

Bases: BaseModel

Runtime REST API connection settings (base URL and credentials).

Source code in graflo/connections/sources.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class RestApiConnConfig(BaseModel):
    """Runtime REST API connection settings (base URL and credentials)."""

    base_url: str
    auth: ApiAuth | None = None
    default_headers: dict[str, str] = Field(default_factory=dict)
    proxy_name: str | None = Field(
        default=None,
        description=(
            "Non-secret runtime handle. Matches conn_proxy in manifest bindings. "
            "Must be unique per registry instance. Mirrors DBConfig.proxy_name: "
            "an API source's connector is looked up by the same label a "
            "database connector is."
        ),
    )

    @classmethod
    def from_env(cls, env_prefix: str) -> RestApiConnConfig:
        """Load REST API config from environment variables.

        Supported variables (all prefixed with *env_prefix*):

        - ``BASE_URL`` (required)
        - ``AUTH_TYPE``: ``bearer``, ``basic``, ``digest``, or ``api_key`` (default: ``bearer``)
        - ``TOKEN``, ``USERNAME``, ``PASSWORD``
        - ``HEADER_NAME``, ``PREFIX`` (bearer / api_key)
        """
        base_url = os.environ.get(f"{env_prefix}BASE_URL")
        if not base_url:
            raise ValueError(
                f"Environment variable {env_prefix}BASE_URL is required for RestApiConnConfig"
            )

        auth_type_raw = os.environ.get(f"{env_prefix}AUTH_TYPE", "bearer")
        auth_type_lower = auth_type_raw.lower()
        if auth_type_lower not in _VALID_AUTH_TYPES:
            raise ValueError(
                f"Invalid {env_prefix}AUTH_TYPE={auth_type_raw!r}; "
                "expected bearer, basic, digest, or api_key"
            )
        auth = ApiAuth(
            auth_type=cast(AuthType, auth_type_lower),
            token=os.environ.get(f"{env_prefix}TOKEN"),
            username=os.environ.get(f"{env_prefix}USERNAME"),
            password=os.environ.get(f"{env_prefix}PASSWORD"),
            header_name=cast(
                str,
                os.environ.get(f"{env_prefix}HEADER_NAME") or "Authorization",
            ),
            prefix=cast(
                str,
                os.environ.get(f"{env_prefix}PREFIX") or "Bearer",
            ),
        )

        return cls(base_url=cast(str, base_url), auth=auth)

Attributes

auth = None class-attribute instance-attribute
base_url instance-attribute
default_headers = Field(default_factory=dict) class-attribute instance-attribute
proxy_name = Field(default=None, description="Non-secret runtime handle. Matches conn_proxy in manifest bindings. Must be unique per registry instance. Mirrors DBConfig.proxy_name: an API source's connector is looked up by the same label a database connector is.") class-attribute instance-attribute

Methods:

from_env(env_prefix) classmethod

Load REST API config from environment variables.

Supported variables (all prefixed with env_prefix):

  • BASE_URL (required)
  • AUTH_TYPE: bearer, basic, digest, or api_key (default: bearer)
  • TOKEN, USERNAME, PASSWORD
  • HEADER_NAME, PREFIX (bearer / api_key)
Source code in graflo/connections/sources.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def from_env(cls, env_prefix: str) -> RestApiConnConfig:
    """Load REST API config from environment variables.

    Supported variables (all prefixed with *env_prefix*):

    - ``BASE_URL`` (required)
    - ``AUTH_TYPE``: ``bearer``, ``basic``, ``digest``, or ``api_key`` (default: ``bearer``)
    - ``TOKEN``, ``USERNAME``, ``PASSWORD``
    - ``HEADER_NAME``, ``PREFIX`` (bearer / api_key)
    """
    base_url = os.environ.get(f"{env_prefix}BASE_URL")
    if not base_url:
        raise ValueError(
            f"Environment variable {env_prefix}BASE_URL is required for RestApiConnConfig"
        )

    auth_type_raw = os.environ.get(f"{env_prefix}AUTH_TYPE", "bearer")
    auth_type_lower = auth_type_raw.lower()
    if auth_type_lower not in _VALID_AUTH_TYPES:
        raise ValueError(
            f"Invalid {env_prefix}AUTH_TYPE={auth_type_raw!r}; "
            "expected bearer, basic, digest, or api_key"
        )
    auth = ApiAuth(
        auth_type=cast(AuthType, auth_type_lower),
        token=os.environ.get(f"{env_prefix}TOKEN"),
        username=os.environ.get(f"{env_prefix}USERNAME"),
        password=os.environ.get(f"{env_prefix}PASSWORD"),
        header_name=cast(
            str,
            os.environ.get(f"{env_prefix}HEADER_NAME") or "Authorization",
        ),
        prefix=cast(
            str,
            os.environ.get(f"{env_prefix}PREFIX") or "Bearer",
        ),
    )

    return cls(base_url=cast(str, base_url), auth=auth)

S3GeneralizedConnConfig

Bases: BaseModel

Runtime credentials and defaults for S3 staging (TigerGraph bulk ingest).

Source code in graflo/connections/sources.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class S3GeneralizedConnConfig(BaseModel):
    """Runtime credentials and defaults for S3 staging (TigerGraph bulk ingest)."""

    kind: Literal["s3"] = "s3"
    bucket: str | None = Field(
        default=None,
        description="Default bucket when TigergraphBulkLoadConfig.s3_bucket is unset.",
    )
    region: str | None = Field(default=None)
    aws_access_key_id: str | None = Field(default=None)
    aws_secret_access_key: str | None = Field(default=None)
    endpoint_url: str | None = Field(
        default=None, description="For S3-compatible endpoints (MinIO, etc.)."
    )
    loader_endpoint_url: str | None = Field(
        default=None,
        description=(
            "S3 endpoint URL as seen by TigerGraph when it runs in another network "
            "namespace (e.g. Docker). Used only in CREATE DATA_SOURCE for LOADING JOB; "
            "boto3 continues to use endpoint_url."
        ),
    )

Attributes

aws_access_key_id = Field(default=None) class-attribute instance-attribute
aws_secret_access_key = Field(default=None) class-attribute instance-attribute
bucket = Field(default=None, description='Default bucket when TigergraphBulkLoadConfig.s3_bucket is unset.') class-attribute instance-attribute
endpoint_url = Field(default=None, description='For S3-compatible endpoints (MinIO, etc.).') class-attribute instance-attribute
kind = 's3' class-attribute instance-attribute
loader_endpoint_url = Field(default=None, description='S3 endpoint URL as seen by TigerGraph when it runs in another network namespace (e.g. Docker). Used only in CREATE DATA_SOURCE for LOADING JOB; boto3 continues to use endpoint_url.') class-attribute instance-attribute
region = Field(default=None) class-attribute instance-attribute

SparqlAuth

Bases: BaseModel

Authentication payload for SPARQL endpoint access.

Source code in graflo/connections/sources.py 
21
22
23
24
25
class SparqlAuth(BaseModel):
    """Authentication payload for SPARQL endpoint access."""

    username: str | None = None
    password: str | None = None

Attributes

password = None class-attribute instance-attribute
username = None class-attribute instance-attribute

SparqlGeneralizedConnConfig

Bases: BaseModel

Generalized runtime config variant for SPARQL endpoint connections.

Source code in graflo/connections/sources.py 
104
105
106
107
108
class SparqlGeneralizedConnConfig(BaseModel):
    """Generalized runtime config variant for SPARQL endpoint connections."""

    kind: Literal["sparql"] = "sparql"
    config: SparqlEndpointConfig

Attributes

config instance-attribute
kind = 'sparql' class-attribute instance-attribute