graflo.connections.provider

Runtime resolution of conn_proxy labels to connection configs.

Manifests stay secret-free: a connector names a conn_proxy label in its bindings, and a :class:ConnectionProvider maps each label to a :data:~graflo.connections.sources.ConnConfig at run time. Which connector uses which label is read from the bindings at lookup, never copied into the provider.

Attributes

ConnConfig = PostgresConfig | SparqlEndpointConfig | RestApiConnConfig | KafkaConnConfig | FilesConnConfig module-attribute

A runtime connection a conn_proxy label resolves to.

__all__ = ['ApiAuth', 'ConnConfig', 'ConnectionProvider', 'EmptyConnectionProvider', 'FilesConnConfig', 'InMemoryConnectionProvider', 'KafkaConnConfig', 'RestApiConnConfig', 'S3ConnConfig', 'SparqlAuth', 'connector_config', 'proxy_env_prefix', 'resolve_db_secrets_from_env', 'resolve_secrets_from_env'] module-attribute

Classes

ApiAuth

Bases: BaseModel

Authentication payload for REST API source access.

Source code in graflo/connections/sources.py

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

ConnectionProvider

Bases: Protocol

Resolve a conn_proxy label to its runtime connection config.

Source code in graflo/connections/provider.py

class ConnectionProvider(Protocol):
    """Resolve a ``conn_proxy`` label to its runtime connection config."""

    def get_config(self, conn_proxy: str) -> ConnConfig | None:
        """The config registered under *conn_proxy*, or ``None``."""
        ...

Methods:

get_config(conn_proxy)

The config registered under conn_proxy, or None.

Source code in graflo/connections/provider.py

def get_config(self, conn_proxy: str) -> ConnConfig | None:
    """The config registered under *conn_proxy*, or ``None``."""
    ...

EmptyConnectionProvider

A provider with no configs: every label resolves to None.

Source code in graflo/connections/provider.py

class EmptyConnectionProvider:
    """A provider with no configs: every label resolves to ``None``."""

    def get_config(self, conn_proxy: str) -> ConnConfig | None:
        return None

Methods:

get_config(conn_proxy)

Source code in graflo/connections/provider.py

def get_config(self, conn_proxy: str) -> ConnConfig | None:
    return None

FilesConnConfig

Bases: BaseModel

A place files are read from: a local directory or an S3 bucket prefix.

root is an absolute local path, or s3://<bucket>[/<prefix>]. A FileConnector bound to this connection reads sub_path relative to the root. The S3 fields apply to an s3:// root only; keys come from {PREFIX}ACCESS_KEY_ID / {PREFIX}SECRET_ACCESS_KEY at run time.

Source code in graflo/connections/sources.py

class FilesConnConfig(BaseModel):
    """A place files are read from: a local directory or an S3 bucket prefix.

    ``root`` is an absolute local path, or ``s3://<bucket>[/<prefix>]``. A
    ``FileConnector`` bound to this connection reads ``sub_path`` relative to
    the root. The S3 fields apply to an ``s3://`` root only; keys come from
    ``{PREFIX}ACCESS_KEY_ID`` / ``{PREFIX}SECRET_ACCESS_KEY`` at run time.
    """

    kind: Literal["files"] = "files"
    root: str = Field(min_length=1)
    endpoint_url: str | None = Field(
        default=None, description="For S3-compatible endpoints (MinIO, SeaweedFS)."
    )
    region: str | None = None
    access_key_id: str | None = None
    secret_access_key: str | None = None
    proxy_name: str | None = Field(default=None, description=_PROXY_NAME_DESCRIPTION)
    alias: str | None = Field(default=None, description=_ALIAS_DESCRIPTION)

    @model_validator(mode="after")
    def _check_root(self) -> FilesConnConfig:
        if self.is_s3:
            if not self.bucket:
                raise ValueError(f"root {self.root!r} names no bucket")
        else:
            if not PurePosixPath(self.root).is_absolute():
                raise ValueError(
                    f"root {self.root!r} must be an absolute path or an s3:// URL"
                )
            s3_only = [
                name
                for name in (
                    "endpoint_url",
                    "region",
                    "access_key_id",
                    "secret_access_key",
                )
                if getattr(self, name) is not None
            ]
            if s3_only:
                raise ValueError(f"{', '.join(s3_only)} apply to an s3:// root only")
        return self

    @property
    def is_s3(self) -> bool:
        return self.root.startswith(_S3_SCHEME)

    @property
    def bucket(self) -> str | None:
        """The bucket of an ``s3://`` root."""
        if not self.is_s3:
            return None
        return self.root[len(_S3_SCHEME) :].split("/", 1)[0] or None

    @property
    def prefix(self) -> str:
        """The key prefix of an ``s3://`` root, without slashes at either end."""
        if not self.is_s3:
            return ""
        _, _, rest = self.root[len(_S3_SCHEME) :].partition("/")
        return rest.strip("/")

    @property
    def local_root(self) -> Path:
        """The directory of a local root."""
        if self.is_s3:
            raise ValueError(f"root {self.root!r} is not a local directory")
        return Path(self.root).expanduser()

    def s3_config(self) -> S3ConnConfig:
        """The boto3 settings of an ``s3://`` root."""
        if not self.is_s3:
            raise ValueError(f"root {self.root!r} is not an s3:// URL")
        return S3ConnConfig(
            bucket=self.bucket,
            region=self.region,
            aws_access_key_id=self.access_key_id,
            aws_secret_access_key=self.secret_access_key,
            endpoint_url=self.endpoint_url,
        )

    @classmethod
    def from_env(cls, env_prefix: str) -> FilesConnConfig:
        """Load from ``{env_prefix}<FIELD>``: ``ROOT`` is required, the rest optional."""
        if not os.environ.get(f"{env_prefix}ROOT"):
            raise ValueError(
                f"Environment variable {env_prefix}ROOT is required for FilesConnConfig"
            )
        return cls.model_validate(
            {
                name: value
                for name in cls.model_fields
                if name != "kind"
                and (value := os.environ.get(f"{env_prefix}{name.upper()}") or None)
                is not None
            }
        )

Attributes

access_key_id = None class-attribute instance-attribute

alias = Field(default=None, description=_ALIAS_DESCRIPTION) class-attribute instance-attribute

bucket property

The bucket of an s3:// root.

endpoint_url = Field(default=None, description='For S3-compatible endpoints (MinIO, SeaweedFS).') class-attribute instance-attribute

is_s3 property

kind = 'files' class-attribute instance-attribute

local_root property

The directory of a local root.

prefix property

The key prefix of an s3:// root, without slashes at either end.

proxy_name = Field(default=None, description=_PROXY_NAME_DESCRIPTION) class-attribute instance-attribute

region = None class-attribute instance-attribute

root = Field(min_length=1) class-attribute instance-attribute

secret_access_key = None class-attribute instance-attribute

Methods:

from_env(env_prefix) classmethod

Load from {env_prefix}<FIELD>: ROOT is required, the rest optional.

Source code in graflo/connections/sources.py

@classmethod
def from_env(cls, env_prefix: str) -> FilesConnConfig:
    """Load from ``{env_prefix}<FIELD>``: ``ROOT`` is required, the rest optional."""
    if not os.environ.get(f"{env_prefix}ROOT"):
        raise ValueError(
            f"Environment variable {env_prefix}ROOT is required for FilesConnConfig"
        )
    return cls.model_validate(
        {
            name: value
            for name in cls.model_fields
            if name != "kind"
            and (value := os.environ.get(f"{env_prefix}{name.upper()}") or None)
            is not None
        }
    )

s3_config()

The boto3 settings of an s3:// root.

Source code in graflo/connections/sources.py

def s3_config(self) -> S3ConnConfig:
    """The boto3 settings of an ``s3://`` root."""
    if not self.is_s3:
        raise ValueError(f"root {self.root!r} is not an s3:// URL")
    return S3ConnConfig(
        bucket=self.bucket,
        region=self.region,
        aws_access_key_id=self.access_key_id,
        aws_secret_access_key=self.secret_access_key,
        endpoint_url=self.endpoint_url,
    )

InMemoryConnectionProvider

conn_proxy label → config, held in memory.

Build it from a mapping, InMemoryConnectionProvider({"shop_db": cfg}), or from the environment with :meth:from_env.

Source code in graflo/connections/provider.py

class InMemoryConnectionProvider:
    """``conn_proxy`` label → config, held in memory.

    Build it from a mapping, ``InMemoryConnectionProvider({"shop_db": cfg})``,
    or from the environment with :meth:`from_env`.
    """

    def __init__(self, configs: Mapping[str, ConnConfig] | None = None) -> None:
        self.configs: dict[str, ConnConfig] = dict(configs or {})

    def register(self, conn_proxy: str, config: ConnConfig) -> None:
        """Register (or replace) the config of *conn_proxy*."""
        self.configs[conn_proxy] = config

    def get_config(self, conn_proxy: str) -> ConnConfig | None:
        return self.configs.get(conn_proxy)

    @classmethod
    def from_env(
        cls,
        bindings: Bindings,
        *,
        env_prefix_map: dict[str, str] | None = None,
    ) -> InMemoryConnectionProvider:
        """One config per ``conn_proxy`` the bindings use, read from the environment.

        The config type follows the connectors that name the label: table →
        :class:`PostgresConfig`, SPARQL → :class:`SparqlEndpointConfig`, REST
        API → :class:`RestApiConnConfig`, Kafka → :class:`KafkaConnConfig`,
        file → :class:`FilesConnConfig`. Each reads ``{PREFIX}<FIELD>``,
        *PREFIX* from :func:`proxy_env_prefix` unless *env_prefix_map* names
        one for the label. Labels named only by push connectors are skipped.

        Raises:
            ValueError: A label is named by connectors needing different config
                types, or a label's required variables are missing — every
                such label is listed in one error.
        """
        prefix_map = env_prefix_map or {}
        configs: dict[str, ConnConfig] = {}
        problems: list[str] = []
        for proxy, connectors in bindings.conn_proxies().items():
            needed = {t for c in connectors if (t := _config_type_for(c)) is not None}
            if not needed:
                continue
            if len(needed) > 1:
                kinds = ", ".join(sorted(t.__name__ for t in needed))
                problems.append(f"{proxy!r} is used by connectors needing {kinds}")
                continue
            config_type = needed.pop()
            prefix = prefix_map.get(proxy, proxy_env_prefix(proxy))
            try:
                config = _load_from_env(config_type, proxy, prefix)
            except ValueError as exc:
                problems.append(f"{proxy!r}: {exc}")
                continue
            configs[proxy] = config
        if problems:
            raise ValueError(
                "Cannot load connection configs from the environment: "
                + "; ".join(problems)
            )
        return cls(configs)

Attributes

configs = dict(configs or {}) instance-attribute

Methods:

__init__(configs=None)

Source code in graflo/connections/provider.py

def __init__(self, configs: Mapping[str, ConnConfig] | None = None) -> None:
    self.configs: dict[str, ConnConfig] = dict(configs or {})

from_env(bindings, *, env_prefix_map=None) classmethod

One config per conn_proxy the bindings use, read from the environment.

The config type follows the connectors that name the label: table → :class:PostgresConfig, SPARQL → :class:SparqlEndpointConfig, REST API → :class:RestApiConnConfig, Kafka → :class:KafkaConnConfig, file → :class:FilesConnConfig. Each reads {PREFIX}<FIELD>, PREFIX from :func:proxy_env_prefix unless env_prefix_map names one for the label. Labels named only by push connectors are skipped.

Raises:

Type	Description
ValueError	A label is named by connectors needing different config types, or a label's required variables are missing — every such label is listed in one error.

Source code in graflo/connections/provider.py

@classmethod
def from_env(
    cls,
    bindings: Bindings,
    *,
    env_prefix_map: dict[str, str] | None = None,
) -> InMemoryConnectionProvider:
    """One config per ``conn_proxy`` the bindings use, read from the environment.

    The config type follows the connectors that name the label: table →
    :class:`PostgresConfig`, SPARQL → :class:`SparqlEndpointConfig`, REST
    API → :class:`RestApiConnConfig`, Kafka → :class:`KafkaConnConfig`,
    file → :class:`FilesConnConfig`. Each reads ``{PREFIX}<FIELD>``,
    *PREFIX* from :func:`proxy_env_prefix` unless *env_prefix_map* names
    one for the label. Labels named only by push connectors are skipped.

    Raises:
        ValueError: A label is named by connectors needing different config
            types, or a label's required variables are missing — every
            such label is listed in one error.
    """
    prefix_map = env_prefix_map or {}
    configs: dict[str, ConnConfig] = {}
    problems: list[str] = []
    for proxy, connectors in bindings.conn_proxies().items():
        needed = {t for c in connectors if (t := _config_type_for(c)) is not None}
        if not needed:
            continue
        if len(needed) > 1:
            kinds = ", ".join(sorted(t.__name__ for t in needed))
            problems.append(f"{proxy!r} is used by connectors needing {kinds}")
            continue
        config_type = needed.pop()
        prefix = prefix_map.get(proxy, proxy_env_prefix(proxy))
        try:
            config = _load_from_env(config_type, proxy, prefix)
        except ValueError as exc:
            problems.append(f"{proxy!r}: {exc}")
            continue
        configs[proxy] = config
    if problems:
        raise ValueError(
            "Cannot load connection configs from the environment: "
            + "; ".join(problems)
        )
    return cls(configs)

get_config(conn_proxy)

Source code in graflo/connections/provider.py

def get_config(self, conn_proxy: str) -> ConnConfig | None:
    return self.configs.get(conn_proxy)

register(conn_proxy, config)

Register (or replace) the config of conn_proxy.

Source code in graflo/connections/provider.py

def register(self, conn_proxy: str, config: ConnConfig) -> None:
    """Register (or replace) the config of *conn_proxy*."""
    self.configs[conn_proxy] = config

KafkaConnConfig

Bases: BaseModel

Runtime Kafka broker connection settings (bootstrap and auth).

Source code in graflo/connections/sources.py

class KafkaConnConfig(BaseModel):
    """Runtime Kafka broker connection settings (bootstrap and auth)."""

    kind: Literal["stream"] = "stream"
    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
    proxy_name: str | None = Field(default=None, description=_PROXY_NAME_DESCRIPTION)
    alias: str | None = Field(default=None, description=_ALIAS_DESCRIPTION)

    @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

alias = Field(default=None, description=_ALIAS_DESCRIPTION) class-attribute instance-attribute

bootstrap_servers instance-attribute

client_id = None class-attribute instance-attribute

kind = 'stream' class-attribute instance-attribute

proxy_name = Field(default=None, description=_PROXY_NAME_DESCRIPTION) 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

@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

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

RestApiConnConfig

Bases: BaseModel

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

Source code in graflo/connections/sources.py

class RestApiConnConfig(BaseModel):
    """Runtime REST API connection settings (base URL and credentials)."""

    kind: Literal["api"] = "api"
    base_url: str
    auth: ApiAuth | None = None
    default_headers: dict[str, str] = Field(default_factory=dict)
    proxy_name: str | None = Field(default=None, description=_PROXY_NAME_DESCRIPTION)
    alias: str | None = Field(default=None, description=_ALIAS_DESCRIPTION)

    @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)

        With none of ``TOKEN``, ``USERNAME`` and ``PASSWORD`` set the API is
        taken to be open: ``auth`` is ``None`` and no credential header is sent.
        """
        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"
            )
        token = os.environ.get(f"{env_prefix}TOKEN")
        username = os.environ.get(f"{env_prefix}USERNAME")
        password = os.environ.get(f"{env_prefix}PASSWORD")
        if not (token or username or password):
            return cls(base_url=cast(str, base_url))

        auth = ApiAuth(
            auth_type=cast(AuthType, auth_type_lower),
            token=token,
            username=username,
            password=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

alias = Field(default=None, description=_ALIAS_DESCRIPTION) class-attribute instance-attribute

auth = None class-attribute instance-attribute

base_url instance-attribute

default_headers = Field(default_factory=dict) class-attribute instance-attribute

kind = 'api' class-attribute instance-attribute

proxy_name = Field(default=None, description=_PROXY_NAME_DESCRIPTION) 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)

With none of TOKEN, USERNAME and PASSWORD set the API is taken to be open: auth is None and no credential header is sent.

Source code in graflo/connections/sources.py

@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)

    With none of ``TOKEN``, ``USERNAME`` and ``PASSWORD`` set the API is
    taken to be open: ``auth`` is ``None`` and no credential header is sent.
    """
    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"
        )
    token = os.environ.get(f"{env_prefix}TOKEN")
    username = os.environ.get(f"{env_prefix}USERNAME")
    password = os.environ.get(f"{env_prefix}PASSWORD")
    if not (token or username or password):
        return cls(base_url=cast(str, base_url))

    auth = ApiAuth(
        auth_type=cast(AuthType, auth_type_lower),
        token=token,
        username=username,
        password=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)

S3ConnConfig

Bases: BaseModel

What a boto3 S3 client needs: endpoint, region and keys, plus a default bucket.

Not a registered connection on its own: a :class:FilesConnConfig with an s3:// root produces one (:meth:FilesConnConfig.s3_config).

Source code in graflo/connections/sources.py

class S3ConnConfig(BaseModel):
    """What a boto3 S3 client needs: endpoint, region and keys, plus a default bucket.

    Not a registered connection on its own: a :class:`FilesConnConfig` with an
    ``s3://`` root produces one (:meth:`FilesConnConfig.s3_config`).
    """

    bucket: str | None = Field(
        default=None,
        description="Default bucket.",
    )
    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.)."
    )

    @classmethod
    def from_env(cls, env_prefix: str) -> S3ConnConfig:
        """Load S3 config from ``{env_prefix}<FIELD>`` variables.

        Reads ``BUCKET``, ``REGION``, ``AWS_ACCESS_KEY_ID``,
        ``AWS_SECRET_ACCESS_KEY`` and ``ENDPOINT_URL``; all are optional.
        """
        return cls(
            **{
                name: value
                for name in cls.model_fields
                if (value := os.environ.get(f"{env_prefix}{name.upper()}") or None)
                is not None
            }
        )

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.') class-attribute instance-attribute

endpoint_url = Field(default=None, description='For S3-compatible endpoints (MinIO, etc.).') class-attribute instance-attribute

region = Field(default=None) class-attribute instance-attribute

Methods:

from_env(env_prefix) classmethod

Load S3 config from {env_prefix}<FIELD> variables.

Reads BUCKET, REGION, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and ENDPOINT_URL; all are optional.

Source code in graflo/connections/sources.py

@classmethod
def from_env(cls, env_prefix: str) -> S3ConnConfig:
    """Load S3 config from ``{env_prefix}<FIELD>`` variables.

    Reads ``BUCKET``, ``REGION``, ``AWS_ACCESS_KEY_ID``,
    ``AWS_SECRET_ACCESS_KEY`` and ``ENDPOINT_URL``; all are optional.
    """
    return cls(
        **{
            name: value
            for name in cls.model_fields
            if (value := os.environ.get(f"{env_prefix}{name.upper()}") or None)
            is not None
        }
    )

SparqlAuth

Bases: BaseModel

Authentication payload for SPARQL endpoint access.

Source code in graflo/connections/sources.py

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

Functions:

connector_config(connector, bindings, provider, expected)

The config of connector's conn_proxy, checked to be an expected.

Raises:

Type	Description
ValueError	The connector has no conn_proxy, nothing is registered under it, or what is registered is another kind of config.

Source code in graflo/connections/provider.py

def connector_config(
    connector: AnyConnector,
    bindings: Bindings,
    provider: ConnectionProvider,
    expected: type[_T],
) -> _T:
    """The config of *connector*'s ``conn_proxy``, checked to be an *expected*.

    Raises:
        ValueError: The connector has no ``conn_proxy``, nothing is registered
            under it, or what is registered is another kind of config.
    """
    label = connector.name or connector.hash
    proxy = bindings.get_conn_proxy_for_connector(connector)
    if proxy is None:
        raise ValueError(
            f"connector {label!r} has no conn_proxy; "
            f"a {expected.__name__} is needed to read it"
        )
    config = provider.get_config(proxy)
    if config is None:
        raise ValueError(
            f"no connection configuration is registered for conn_proxy "
            f"{proxy!r} (connector {label!r})"
        )
    if not isinstance(config, expected):
        raise ValueError(
            f"conn_proxy {proxy!r} (connector {label!r}) holds a "
            f"{type(config).__name__}; a {expected.__name__} is needed"
        )
    return config

proxy_env_prefix(conn_proxy)

The environment-variable prefix of a conn_proxy label.

user_service → USER_SERVICE_; billing-api → BILLING_API_.

Source code in graflo/connections/env.py

def proxy_env_prefix(conn_proxy: str) -> str:
    """The environment-variable prefix of a ``conn_proxy`` label.

    ``user_service`` → ``USER_SERVICE_``; ``billing-api`` → ``BILLING_API_``.
    """
    return conn_proxy.upper().replace("-", "_") + "_"

resolve_db_secrets_from_env(config, conn_proxy, env_prefix=None)

Fill a stored database config's missing username and password from the environment.

The same rule as :func:resolve_secrets_from_env, for any database config, a graph target included: {prefix}USERNAME / {prefix}PASSWORD fill what config lacks and never override a stored value.

Returns:

Type	Description
_D	A copy of config with missing credentials filled, or config itself
_D	when nothing was filled.

Source code in graflo/connections/provider.py

def resolve_db_secrets_from_env(
    config: _D,
    conn_proxy: str,
    env_prefix: str | None = None,
) -> _D:
    """Fill a stored database config's missing username and password from the environment.

    The same rule as :func:`resolve_secrets_from_env`, for any database config,
    a graph target included: ``{prefix}USERNAME`` / ``{prefix}PASSWORD`` fill
    what *config* lacks and never override a stored value.

    Returns:
        A copy of *config* with missing credentials filled, or *config* itself
        when nothing was filled.
    """
    prefix = env_prefix if env_prefix is not None else proxy_env_prefix(conn_proxy)
    update = {
        field: value
        for field in ("username", "password")
        if (value := os.environ.get(f"{prefix}{field.upper()}") or None) is not None
        and getattr(config, field) is None
    }
    return config.model_copy(update=update) if update else config

resolve_secrets_from_env(config, conn_proxy, env_prefix=None)

Fill a stored config's missing credentials from the environment.

A registry stores connections without secrets; the secret behind a conn_proxy label arrives at run time, read from {prefix}<NAME> where prefix defaults to :func:proxy_env_prefix:

• database and SPARQL: USERNAME, PASSWORD;
• REST API: TOKEN, USERNAME, PASSWORD;
• Kafka: SASL_USERNAME, SASL_PASSWORD;
• files on S3: ACCESS_KEY_ID, SECRET_ACCESS_KEY.

A value already present on config is kept: the environment fills gaps and never overrides a stored setting.

Returns:

Type	Description
_C	A copy of config with missing credentials filled, or config itself
_C	when nothing was filled.

Source code in graflo/connections/provider.py

def resolve_secrets_from_env(
    config: _C,
    conn_proxy: str,
    env_prefix: str | None = None,
) -> _C:
    """Fill a stored config's missing credentials from the environment.

    A registry stores connections without secrets; the secret behind a
    ``conn_proxy`` label arrives at run time, read from ``{prefix}<NAME>``
    where *prefix* defaults to :func:`proxy_env_prefix`:

    - database and SPARQL: ``USERNAME``, ``PASSWORD``;
    - REST API: ``TOKEN``, ``USERNAME``, ``PASSWORD``;
    - Kafka: ``SASL_USERNAME``, ``SASL_PASSWORD``;
    - files on S3: ``ACCESS_KEY_ID``, ``SECRET_ACCESS_KEY``.

    A value already present on *config* is kept: the environment fills gaps
    and never overrides a stored setting.

    Returns:
        A copy of *config* with missing credentials filled, or *config* itself
        when nothing was filled.
    """
    prefix = env_prefix if env_prefix is not None else proxy_env_prefix(conn_proxy)

    if isinstance(config, DBConfig):
        return resolve_db_secrets_from_env(config, conn_proxy, prefix)

    if isinstance(config, KafkaConnConfig):
        return cast(
            _C, _fill_from_env(config, ("sasl_username", "sasl_password"), prefix)
        )

    if isinstance(config, FilesConnConfig):
        if not config.is_s3:
            return config
        return cast(
            _C,
            _fill_from_env(config, ("access_key_id", "secret_access_key"), prefix),
        )

    if isinstance(config, RestApiConnConfig):
        auth = config.auth or ApiAuth()
        update = {
            field: value
            for field in ("token", "username", "password")
            if (value := os.environ.get(f"{prefix}{field.upper()}") or None) is not None
            and getattr(auth, field) is None
        }
        if not update:
            return config
        return config.model_copy(update={"auth": auth.model_copy(update=update)})

    return config