graflo.connections.kinds

Every kind of connection as one typed union, and the roles each kind plays.

A connection is a database, a REST API, a stream or a place files live. Each config carries a kind (and a database its db_type), so a stored connection always parses back to the same class, and what it can be used for follows from the config alone, without contacting anything.

Attributes

AnyConnection = DBConfig | RestApiConnConfig | KafkaConnConfig | FilesConnConfig module-attribute

The runtime type of a parsed :data:Connection.

CONNECTION_ADAPTER = TypeAdapter(Connection) module-attribute

Connection = Annotated[DatabaseConnection | RestApiConnConfig | KafkaConnConfig | FilesConnConfig, Field(discriminator='kind')] module-attribute

Any connection, discriminated by kind (a database further by db_type).

DatabaseConnection = Annotated[Annotated[ArangoConfig, Tag('arango')] | Annotated[Neo4jConfig, Tag('neo4j')] | Annotated[TigergraphConfig, Tag('tigergraph')] | Annotated[FalkordbConfig, Tag('falkordb')] | Annotated[MemgraphConfig, Tag('memgraph')] | Annotated[NebulaConfig, Tag('nebula')] | Annotated[GraFloBackendConfig, Tag('graflo_backend')] | Annotated[PostgresConfig, Tag('postgres')] | Annotated[SparqlEndpointConfig, Tag('sparql')] | Annotated[GraphDBConfig, Tag('graphdb')], Discriminator(_database_tag)] module-attribute

A database connection, discriminated by db_type.

__all__ = ['CONNECTION_ADAPTER', 'AnyConnection', 'Connection', 'ConnectionKind', 'DatabaseConnection', 'Role', 'connection_kind', 'parse_connection', 'roles'] module-attribute

Classes

ConnectionKind

Bases: StrEnum

What a connection reaches.

Source code in graflo/connections/kinds.py

class ConnectionKind(StrEnum):
    """What a connection reaches."""

    DATABASE = "database"
    API = "api"
    STREAM = "stream"
    FILES = "files"

Attributes

API = 'api' class-attribute instance-attribute

DATABASE = 'database' class-attribute instance-attribute

FILES = 'files' class-attribute instance-attribute

STREAM = 'stream' class-attribute instance-attribute

Role

Bases: StrEnum

What a connection is used as: read from, or written to.

Source code in graflo/connections/kinds.py

class Role(StrEnum):
    """What a connection is used as: read from, or written to."""

    SOURCE = "source"
    TARGET = "target"

Attributes

SOURCE = 'source' class-attribute instance-attribute

TARGET = 'target' class-attribute instance-attribute

Functions:

connection_kind(conn)

The kind of conn.

Source code in graflo/connections/kinds.py

def connection_kind(conn: AnyConnection) -> ConnectionKind:
    """The kind of *conn*."""
    return ConnectionKind(conn.kind)

parse_connection(data)

The connection data describes.

kind is required. A database also needs db_type and is read by :meth:DBConfig.from_dict, which accepts its URI spelt in parts.

Raises:

Type	Description
ValueError	The kind is missing or unknown, or the config does not validate for it.

Source code in graflo/connections/kinds.py

def parse_connection(data: dict[str, Any]) -> AnyConnection:
    """The connection *data* describes.

    ``kind`` is required. A database also needs ``db_type`` and is read by
    :meth:`DBConfig.from_dict`, which accepts its URI spelt in parts.

    Raises:
        ValueError: The kind is missing or unknown, or the config does not
            validate for it.
    """
    kind = data.get("kind")
    if kind is None:
        raise ValueError(
            "a connection needs a 'kind': one of "
            + ", ".join(k.value for k in ConnectionKind)
        )
    if kind == ConnectionKind.DATABASE:
        return DBConfig.from_dict(data)
    config_class = _CONFIG_BY_KIND.get(str(kind))
    if config_class is None:
        raise ValueError(
            f"unknown connection kind {kind!r}; expected one of "
            + ", ".join(k.value for k in ConnectionKind)
        )
    return config_class.model_validate(data)

roles(conn)

What conn can be used as, from its kind and flavour alone.

An API, a stream and a files connection are read from. A database is read from when a connector reads its rows (:data:SOURCE_DATABASES) and written to when it holds a graph (:data:TARGET_DATABASES); PostgreSQL is both.

Source code in graflo/connections/kinds.py

def roles(conn: AnyConnection) -> frozenset[Role]:
    """What *conn* can be used as, from its kind and flavour alone.

    An API, a stream and a files connection are read from. A database is read
    from when a connector reads its rows (:data:`SOURCE_DATABASES`) and written
    to when it holds a graph (:data:`TARGET_DATABASES`); PostgreSQL is both.
    """
    if not isinstance(conn, DBConfig):
        return frozenset({Role.SOURCE})
    found: set[Role] = set()
    if conn.db_type in SOURCE_DATABASES:
        found.add(Role.SOURCE)
    if conn.db_type in TARGET_DATABASES:
        found.add(Role.TARGET)
    return frozenset(found)