graflo.hq.discovery

What a connection offers to read, before any model exists.

:func:discover turns one connection into candidate connectors: the tables of a database, the classes of a SPARQL endpoint, the topics of a Kafka cluster, the files under a files connection. A pick of them becomes a bindings block (:meth:Discovery.bindings), sampled and ingested like any other.

Files in one directory whose names differ only in runs of digits (part-0001.csv, part-0002.csv) are one candidate, read by a connector whose regex lets those runs vary; a local directory and an S3 prefix are listed alike.

A REST API cannot list its own endpoints, so its discovery is empty: its connectors come from the models that bind them.

Attributes

DEFAULT_DISCOVERY_LIMIT = 500 module-attribute

KAFKA_METADATA_TIMEOUT_S = 10.0 module-attribute

__all__ = ['DEFAULT_DISCOVERY_LIMIT', 'DiscoveredColumn', 'DiscoveredResource', 'Discovery', 'bindings_for', 'discover', 'kafka_topics', 'sparql_select'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

DiscoveredColumn

Bases: ConfigBaseModel

A column a table declares.

Source code in graflo/hq/discovery.py

class DiscoveredColumn(ConfigBaseModel):
    """A column a table declares."""

    name: str
    type: str
    description: str | None = None
    is_pk: bool = False
    is_unique: bool = False
    nullable: bool = True

Attributes

description = None class-attribute instance-attribute

is_pk = False class-attribute instance-attribute

is_unique = False class-attribute instance-attribute

name instance-attribute

nullable = True class-attribute instance-attribute

type instance-attribute

DiscoveredResource

Bases: ConfigBaseModel

One thing a connection offers to read, as the connector that reads it.

Attributes:

Name	Type	Description
name	str	Resource name the pick is sampled under; unique in its discovery.
connector	AnyConnector	The connector that reads it; its name is name.
row_estimate	int | None	Approximate rows (tables) or instances (SPARQL classes).
size_bytes	int | None	Total size of the files a file connector matches.
file_count	int | None	Files a file connector matches.
description	str | None	What the source says the thing is (a table comment).
columns,	(primary_key, foreign_keys)	What a table declares.

Source code in graflo/hq/discovery.py

class DiscoveredResource(ConfigBaseModel):
    """One thing a connection offers to read, as the connector that reads it.

    Attributes:
        name: Resource name the pick is sampled under; unique in its discovery.
        connector: The connector that reads it; its ``name`` is ``name``.
        row_estimate: Approximate rows (tables) or instances (SPARQL classes).
        size_bytes: Total size of the files a file connector matches.
        file_count: Files a file connector matches.
        description: What the source says the thing is (a table comment).
        columns, primary_key, foreign_keys: What a table declares.
    """

    name: str
    connector: AnyConnector
    row_estimate: int | None = None
    size_bytes: int | None = None
    file_count: int | None = None
    description: str | None = None
    columns: list[DiscoveredColumn] = Field(default_factory=list)
    primary_key: list[str] = Field(default_factory=list)
    foreign_keys: list[ForeignKeyHint] = Field(default_factory=list)

Attributes

columns = Field(default_factory=list) class-attribute instance-attribute

connector instance-attribute

description = None class-attribute instance-attribute

file_count = None class-attribute instance-attribute

foreign_keys = Field(default_factory=list) class-attribute instance-attribute

name instance-attribute

primary_key = Field(default_factory=list) class-attribute instance-attribute

row_estimate = None class-attribute instance-attribute

size_bytes = None class-attribute instance-attribute

Discovery

Bases: ConfigBaseModel

The candidates one connection offers.

Attributes:

Name	Type	Description
kind	ConnectionKind	Kind of the connection discovered.
resources	list[DiscoveredResource]	Candidates, in the source's order.
truncated	bool	More candidates exist than were listed.

Source code in graflo/hq/discovery.py

class Discovery(ConfigBaseModel):
    """The candidates one connection offers.

    Attributes:
        kind: Kind of the connection discovered.
        resources: Candidates, in the source's order.
        truncated: More candidates exist than were listed.
    """

    kind: ConnectionKind
    resources: list[DiscoveredResource] = Field(default_factory=list)
    truncated: bool = False

    def bindings(self, conn_proxy: str, names: list[str] | None = None) -> Bindings:
        """A bindings block reading the picked candidates through *conn_proxy*.

        See :func:`bindings_for`.

        Raises:
            ValueError: A name in *names* is not a candidate.
        """
        by_name = {resource.name: resource for resource in self.resources}
        wanted = list(by_name) if names is None else names
        unknown = [name for name in wanted if name not in by_name]
        if unknown:
            raise ValueError(f"not discovered here: {', '.join(unknown)}")
        return bindings_for(conn_proxy, [by_name[name] for name in wanted])

Attributes

kind instance-attribute

resources = Field(default_factory=list) class-attribute instance-attribute

truncated = False class-attribute instance-attribute

Methods:

bindings(conn_proxy, names=None)

A bindings block reading the picked candidates through conn_proxy.

See :func:bindings_for.

Raises:

Type	Description
ValueError	A name in names is not a candidate.

Source code in graflo/hq/discovery.py

def bindings(self, conn_proxy: str, names: list[str] | None = None) -> Bindings:
    """A bindings block reading the picked candidates through *conn_proxy*.

    See :func:`bindings_for`.

    Raises:
        ValueError: A name in *names* is not a candidate.
    """
    by_name = {resource.name: resource for resource in self.resources}
    wanted = list(by_name) if names is None else names
    unknown = [name for name in wanted if name not in by_name]
    if unknown:
        raise ValueError(f"not discovered here: {', '.join(unknown)}")
    return bindings_for(conn_proxy, [by_name[name] for name in wanted])

Functions:

bindings_for(conn_proxy, resources)

A bindings block reading resources through the connection conn_proxy names.

Each resource is read by its connector, renamed to the resource's name so connector names are unique when resource names are, and wired to conn_proxy.

Source code in graflo/hq/discovery.py

def bindings_for(conn_proxy: str, resources: Iterable[DiscoveredResource]) -> Bindings:
    """A bindings block reading *resources* through the connection *conn_proxy* names.

    Each resource is read by its connector, renamed to the resource's name so
    connector names are unique when resource names are, and wired to
    *conn_proxy*.
    """
    picked = list(resources)
    return Bindings(
        connectors=[
            resource.connector.model_copy(update={"name": resource.name})
            for resource in picked
        ],
        resource_connector=[
            {"resource": resource.name, "connector": resource.name}
            for resource in picked
        ],
        connector_connection=[
            {"connector": resource.name, "conn_proxy": conn_proxy}
            for resource in picked
        ],
    )

discover(conn, *, limit=DEFAULT_DISCOVERY_LIMIT)

The candidates conn offers to read, at most limit of them.

Raises:

Type	Description
ValueError	conn is not read from (a graph database), or is a database flavour no connector reads.

Source code in graflo/hq/discovery.py

def discover(conn: AnyConnection, *, limit: int = DEFAULT_DISCOVERY_LIMIT) -> Discovery:
    """The candidates *conn* offers to read, at most *limit* of them.

    Raises:
        ValueError: *conn* is not read from (a graph database), or is a
            database flavour no connector reads.
    """
    if Role.SOURCE not in roles(conn):
        raise ValueError(
            f"a {getattr(conn, 'db_type', connection_kind(conn))} connection is "
            "written to, not read from"
        )
    if isinstance(conn, PostgresConfig):
        return _discover_postgres(conn, limit)
    if isinstance(conn, SparqlEndpointConfig):
        return _discover_sparql(conn, limit)
    if isinstance(conn, KafkaConnConfig):
        return _discover_kafka(conn, limit)
    if isinstance(conn, FilesConnConfig):
        return _discover_files(conn, limit)
    if isinstance(conn, RestApiConnConfig):
        return Discovery(kind=ConnectionKind.API)
    raise ValueError(f"nothing discovers a {type(conn).__name__}")

kafka_topics(conn)

The topics conn's cluster holds, internal ones (__) left out.

Source code in graflo/hq/discovery.py

def kafka_topics(conn: KafkaConnConfig) -> list[str]:
    """The topics *conn*'s cluster holds, internal ones (``__``) left out."""
    from confluent_kafka.admin import AdminClient

    admin = AdminClient(dict[str, Any](conn.to_consumer_config()))
    metadata = admin.list_topics(timeout=KAFKA_METADATA_TIMEOUT_S)
    return sorted(topic for topic in metadata.topics if not topic.startswith("__"))

sparql_select(conn, query)

The bindings of a SELECT query against conn's endpoint.

Source code in graflo/hq/discovery.py

def sparql_select(conn: SparqlEndpointConfig, query: str) -> list[dict[str, Any]]:
    """The bindings of a ``SELECT`` *query* against *conn*'s endpoint."""
    from SPARQLWrapper import JSON, SPARQLWrapper

    client = SPARQLWrapper(conn.query_endpoint)
    client.setQuery(query)
    client.setReturnFormat(JSON)
    if conn.username and conn.password:
        client.setCredentials(conn.username, conn.password)
    result: Any = client.query().convert()
    return list(result.get("results", {}).get("bindings", []))