Skip to content

graflo.db.sql.provider

What a relational source must answer to be introspected.

Introspection needs seven facts about a schema. Everything downstream — classifying a table as an entity or a relationship, recovering edge endpoints, mapping columns onto typed fields — is derived from those and is the same for every dialect. Naming them as a protocol is what lets PostgreSQL keep its pg_catalog fast path while any other engine arrives through SQLAlchemy reflection, with one implementation of the logic between them.

Classes

SqlMetadataProvider

Bases: Protocol

Schema metadata for one relational database.

schema_name is the namespace to read: a PostgreSQL schema, a MySQL database, a BigQuery dataset. Implementations accept None and fall back to their own default.

Source code in graflo/db/sql/provider.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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
@runtime_checkable
class SqlMetadataProvider(Protocol):
    """Schema metadata for one relational database.

    ``schema_name`` is the namespace to read: a PostgreSQL schema, a MySQL
    database, a BigQuery dataset. Implementations accept ``None`` and fall back
    to their own default.
    """

    def get_tables(self, schema_name: str | None = None) -> list[dict[str, Any]]:
        """Tables in *schema_name*, each as a dict with at least ``table_name``."""
        ...

    def get_table_columns(
        self, table_name: str, schema_name: str | None = None
    ) -> list[dict[str, Any]]:
        """Columns of *table_name*.

        Each dict carries ``name`` and ``type``, and may carry ``description``,
        ``is_nullable``, ``column_default`` and ``ordinal_position``.
        """
        ...

    def get_primary_keys(
        self, table_name: str, schema_name: str | None = None
    ) -> list[str]:
        """Primary-key column names, in key order."""
        ...

    def get_unique_columns(
        self, table_name: str, schema_name: str | None = None
    ) -> list[str]:
        """Columns carrying a single-column uniqueness constraint."""
        ...

    def get_foreign_keys(
        self, table_name: str, schema_name: str | None = None
    ) -> list[dict[str, Any]]:
        """Foreign keys, each with ``column`` and ``references_table``.

        ``references_column`` and ``constraint_name`` are optional. An engine
        that does not enforce foreign keys may still declare them (BigQuery), and
        one that enforces them may have none declared — in which case edge
        detection falls back to name-based inference.
        """
        ...

    def get_table_row_count_estimate(
        self, table_name: str, schema_name: str | None = None
    ) -> int | None:
        """Approximate row count, or ``None`` when the engine cannot cheaply say."""
        ...

    def get_table_sample_rows(
        self, table_name: str, schema_name: str | None = None, limit: int = 5
    ) -> list[dict[str, Any]]:
        """A few rows, for column sampling, in a stable order where the engine
        allows (primary key first). Empty list when unavailable."""
        ...

Methods:

get_foreign_keys(table_name, schema_name=None)

Foreign keys, each with column and references_table.

references_column and constraint_name are optional. An engine that does not enforce foreign keys may still declare them (BigQuery), and one that enforces them may have none declared — in which case edge detection falls back to name-based inference.

Source code in graflo/db/sql/provider.py 
51
52
53
54
55
56
57
58
59
60
61
def get_foreign_keys(
    self, table_name: str, schema_name: str | None = None
) -> list[dict[str, Any]]:
    """Foreign keys, each with ``column`` and ``references_table``.

    ``references_column`` and ``constraint_name`` are optional. An engine
    that does not enforce foreign keys may still declare them (BigQuery), and
    one that enforces them may have none declared — in which case edge
    detection falls back to name-based inference.
    """
    ...
get_primary_keys(table_name, schema_name=None)

Primary-key column names, in key order.

Source code in graflo/db/sql/provider.py 
39
40
41
42
43
def get_primary_keys(
    self, table_name: str, schema_name: str | None = None
) -> list[str]:
    """Primary-key column names, in key order."""
    ...
get_table_columns(table_name, schema_name=None)

Columns of table_name.

Each dict carries name and type, and may carry description, is_nullable, column_default and ordinal_position.

Source code in graflo/db/sql/provider.py 
29
30
31
32
33
34
35
36
37
def get_table_columns(
    self, table_name: str, schema_name: str | None = None
) -> list[dict[str, Any]]:
    """Columns of *table_name*.

    Each dict carries ``name`` and ``type``, and may carry ``description``,
    ``is_nullable``, ``column_default`` and ``ordinal_position``.
    """
    ...
get_table_row_count_estimate(table_name, schema_name=None)

Approximate row count, or None when the engine cannot cheaply say.

Source code in graflo/db/sql/provider.py 
63
64
65
66
67
def get_table_row_count_estimate(
    self, table_name: str, schema_name: str | None = None
) -> int | None:
    """Approximate row count, or ``None`` when the engine cannot cheaply say."""
    ...
get_table_sample_rows(table_name, schema_name=None, limit=5)

A few rows, for column sampling, in a stable order where the engine allows (primary key first). Empty list when unavailable.

Source code in graflo/db/sql/provider.py 
69
70
71
72
73
74
def get_table_sample_rows(
    self, table_name: str, schema_name: str | None = None, limit: int = 5
) -> list[dict[str, Any]]:
    """A few rows, for column sampling, in a stable order where the engine
    allows (primary key first). Empty list when unavailable."""
    ...
get_tables(schema_name=None)

Tables in schema_name, each as a dict with at least table_name.

Source code in graflo/db/sql/provider.py 
25
26
27
def get_tables(self, schema_name: str | None = None) -> list[dict[str, Any]]:
    """Tables in *schema_name*, each as a dict with at least ``table_name``."""
    ...
get_unique_columns(table_name, schema_name=None)

Columns carrying a single-column uniqueness constraint.

Source code in graflo/db/sql/provider.py 
45
46
47
48
49
def get_unique_columns(
    self, table_name: str, schema_name: str | None = None
) -> list[str]:
    """Columns carrying a single-column uniqueness constraint."""
    ...