graflo.architecture.onto_sample

Resource sampling contract — pure-JSON samples and their derived profiles.

Samples are the raw material every schema inferencer consumes, whether it reasons algorithmically (:mod:graflo.db.identity_inference) or with a language model (ScheWea). Two ideas, deliberately kept apart:

• Sampling pulls documents from a connector. :class:ResourceSample holds them verbatim as JSON — tabular sources yield flat list[dict] rows, API sources yield arbitrarily nested documents. Nothing is flattened at this boundary, so a hierarchical response survives intact.
• Profiling describes those documents. :func:profile_sample derives the flat, path-keyed, typed view (:class:ResourceProfile) used for prompting, studio previews and identity inference.

:attr:ResourceSample.connector records where the documents came from. That relation is what later becomes a resource plus its resource_connector binding, so it must survive the round trip — a sample that has lost its provenance cannot be turned back into an ingestion model.

Attributes

DEFAULT_MAX_DOCS = 100 module-attribute

DEFAULT_MAX_EXAMPLES = 3 module-attribute

DEFAULT_MAX_PATHS = 200 module-attribute

LIST_MARKER = '[]' module-attribute

__all__ = ['DEFAULT_MAX_DOCS', 'DEFAULT_MAX_EXAMPLES', 'DEFAULT_MAX_PATHS', 'LIST_MARKER', 'FieldProfile', 'ForeignKeyHint', 'ResourceProfile', 'ResourceSample', 'SourceSample', 'infer_field_type', 'iter_paths', 'profile_sample', 'profile_source'] module-attribute

Classes

FieldProfile

Bases: ConfigBaseModel

Derived description of one field path within a resource sample.

Source code in graflo/architecture/onto_sample.py

class FieldProfile(ConfigBaseModel):
    """Derived description of one field path within a resource sample."""

    path: str
    """Dotted path to the value, e.g. ``address.city`` or ``items[].sku``."""

    type: FieldType = FieldType.STRING
    item_type: FieldType | None = None
    """Element type; set only when ``type`` is ``LIST``."""

    depth: int = 0
    """Nesting depth of the path. ``0`` for a top-level scalar."""

    present: int = 0
    """Documents in which the path occurred."""

    null_count: int = 0
    distinct: int = 0
    examples: list[str] = PydanticField(default_factory=list)
    description: str | None = None
    """The source's own description of the field, when it declares one."""

    @property
    def null_ratio(self) -> float:
        """Fraction of occurrences whose value was null."""
        return self.null_count / self.present if self.present else 0.0

    @property
    def unique(self) -> bool:
        """True when every non-null occurrence was distinct."""
        non_null = self.present - self.null_count
        return non_null > 0 and self.distinct == non_null

Attributes

depth = 0 class-attribute instance-attribute

Nesting depth of the path. 0 for a top-level scalar.

description = None class-attribute instance-attribute

The source's own description of the field, when it declares one.

distinct = 0 class-attribute instance-attribute

examples = PydanticField(default_factory=list) class-attribute instance-attribute

item_type = None class-attribute instance-attribute

Element type; set only when type is LIST.

null_count = 0 class-attribute instance-attribute

null_ratio property

Fraction of occurrences whose value was null.

path instance-attribute

Dotted path to the value, e.g. address.city or items[].sku.

present = 0 class-attribute instance-attribute

Documents in which the path occurred.

type = FieldType.STRING class-attribute instance-attribute

unique property

True when every non-null occurrence was distinct.

ForeignKeyHint

Bases: ConfigBaseModel

A declared reference from one resource to another.

Populated only when the source declares it (a SQL foreign key, an RDF range). This is ground truth for edge inference and must not be confused with the name-suffix guessing an inferencer falls back to.

Source code in graflo/architecture/onto_sample.py

class ForeignKeyHint(ConfigBaseModel):
    """A declared reference from one resource to another.

    Populated only when the source *declares* it (a SQL foreign key, an RDF
    range). This is ground truth for edge inference and must not be confused
    with the name-suffix guessing an inferencer falls back to.
    """

    field: str
    references_resource: str
    references_field: str | None = None

Attributes

field instance-attribute

references_field = None class-attribute instance-attribute

references_resource instance-attribute

ResourceProfile

Bases: ConfigBaseModel

Derived, flat, typed view of a :class:ResourceSample.

Source code in graflo/architecture/onto_sample.py

class ResourceProfile(ConfigBaseModel):
    """Derived, flat, typed view of a :class:`ResourceSample`."""

    resource_name: str
    connector: str | None = None
    description: str | None = None
    """Carried from :attr:`ResourceSample.description`."""

    doc_count: int = 0
    max_depth: int = 0
    """Deepest nesting observed. ``> 0`` means ingestion needs ``descend`` steps."""

    fields: list[FieldProfile] = PydanticField(default_factory=list)
    primary_key: list[str] = PydanticField(default_factory=list)
    foreign_keys: list[ForeignKeyHint] = PydanticField(default_factory=list)
    truncated: bool = False

    @property
    def field_paths(self) -> list[str]:
        return [field.path for field in self.fields]

    @property
    def nested(self) -> bool:
        return self.max_depth > 0

    def flat_docs(self, docs: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """Project *docs* onto this profile's paths as flat records.

        Identity inference operates on flat records; this is how a nested source
        becomes eligible for it.
        """
        paths = set(self.field_paths)
        flattened: list[dict[str, Any]] = []
        for doc in docs:
            record: dict[str, Any] = {}
            for path, _depth, value in iter_paths(doc):
                if path in paths and path not in record:
                    record[path] = value
            flattened.append(record)
        return flattened

Attributes

connector = None class-attribute instance-attribute

description = None class-attribute instance-attribute

Carried from :attr:ResourceSample.description.

doc_count = 0 class-attribute instance-attribute

field_paths property

fields = PydanticField(default_factory=list) class-attribute instance-attribute

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

max_depth = 0 class-attribute instance-attribute

Deepest nesting observed. > 0 means ingestion needs descend steps.

nested property

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

resource_name instance-attribute

truncated = False class-attribute instance-attribute

Methods:

flat_docs(docs)

Project docs onto this profile's paths as flat records.

Identity inference operates on flat records; this is how a nested source becomes eligible for it.

Source code in graflo/architecture/onto_sample.py

def flat_docs(self, docs: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Project *docs* onto this profile's paths as flat records.

    Identity inference operates on flat records; this is how a nested source
    becomes eligible for it.
    """
    paths = set(self.field_paths)
    flattened: list[dict[str, Any]] = []
    for doc in docs:
        record: dict[str, Any] = {}
        for path, _depth, value in iter_paths(doc):
            if path in paths and path not in record:
                record[path] = value
        flattened.append(record)
    return flattened

ResourceSample

Bases: ConfigBaseModel

Documents sampled from one resource, plus what the source declared about it.

Source code in graflo/architecture/onto_sample.py

class ResourceSample(ConfigBaseModel):
    """Documents sampled from one resource, plus what the source declared about it."""

    resource_name: str
    """Logical resource name; becomes ``ResourceConfig.name``."""

    connector: str | None = None
    """Name of the connector the documents came from; becomes the
    ``resource_connector`` binding. ``None`` when sampled without bindings."""

    docs: list[dict[str, Any]] = PydanticField(default_factory=list)
    """Sampled documents, verbatim JSON. Flat rows for tables, nested for APIs."""

    description: str | None = None
    """What the source says the resource is -- a SQL table comment, say."""

    field_descriptions: dict[str, str] = PydanticField(default_factory=dict)
    """What the source says each field is, keyed by field path -- SQL column
    comments, for instance. Only declared text; never inferred."""

    primary_key: list[str] = PydanticField(default_factory=list)
    """Declared primary key, when the source has one."""

    foreign_keys: list[ForeignKeyHint] = PydanticField(default_factory=list)
    """Declared outbound references, when the source has them."""

    truncated: bool = False
    """True when documents were dropped or values clipped to respect caps."""

    total_estimate: int | None = None
    """Approximate total document count at the source, when cheaply available."""

Attributes

connector = None class-attribute instance-attribute

Name of the connector the documents came from; becomes the resource_connector binding. None when sampled without bindings.

description = None class-attribute instance-attribute

What the source says the resource is -- a SQL table comment, say.

docs = PydanticField(default_factory=list) class-attribute instance-attribute

Sampled documents, verbatim JSON. Flat rows for tables, nested for APIs.

field_descriptions = PydanticField(default_factory=dict) class-attribute instance-attribute

What the source says each field is, keyed by field path -- SQL column comments, for instance. Only declared text; never inferred.

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

Declared outbound references, when the source has them.

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

Declared primary key, when the source has one.

resource_name instance-attribute

Logical resource name; becomes ResourceConfig.name.

total_estimate = None class-attribute instance-attribute

Approximate total document count at the source, when cheaply available.

truncated = False class-attribute instance-attribute

True when documents were dropped or values clipped to respect caps.

SourceSample

Bases: ConfigBaseModel

A set of resource samples drawn from one logical source.

Source code in graflo/architecture/onto_sample.py

class SourceSample(ConfigBaseModel):
    """A set of resource samples drawn from one logical source."""

    source_name: str
    description: str | None = None
    samples: list[ResourceSample] = PydanticField(min_length=1)

    @model_validator(mode="after")
    def _require_unique_resource_names(self) -> SourceSample:
        """Reject duplicate resource names.

        ``samples_by_resource`` keys on the name, so a duplicate would silently
        discard every sample but the last — losing documents without a trace.
        """
        seen: set[str] = set()
        duplicates: set[str] = set()
        for sample in self.samples:
            if sample.resource_name in seen:
                duplicates.add(sample.resource_name)
            seen.add(sample.resource_name)
        if duplicates:
            raise ValueError(
                f"SourceSample '{self.source_name}': duplicate resource names "
                f"{sorted(duplicates)}. Merge their documents into one sample — "
                "keying by name would drop all but the last."
            )
        return self

    @property
    def samples_by_resource(self) -> dict[str, list[dict[str, Any]]]:
        """Documents keyed by resource name.

        This is the input shape consumed by cross-resource identity inference,
        so no adapter is needed between sampling and inference.

        Returns the *live* document lists, not copies — treat them as read-only.
        """
        return {sample.resource_name: sample.docs for sample in self.samples}

    def get(self, resource_name: str) -> ResourceSample | None:
        """Return the sample for *resource_name*, or ``None``."""
        for sample in self.samples:
            if sample.resource_name == resource_name:
                return sample
        return None

Attributes

description = None class-attribute instance-attribute

samples = PydanticField(min_length=1) class-attribute instance-attribute

samples_by_resource property

Documents keyed by resource name.

This is the input shape consumed by cross-resource identity inference, so no adapter is needed between sampling and inference.

Returns the live document lists, not copies — treat them as read-only.

source_name instance-attribute

Methods:

get(resource_name)

Return the sample for resource_name, or None.

Source code in graflo/architecture/onto_sample.py

def get(self, resource_name: str) -> ResourceSample | None:
    """Return the sample for *resource_name*, or ``None``."""
    for sample in self.samples:
        if sample.resource_name == resource_name:
            return sample
    return None

Functions:

infer_field_type(values)

Infer a FieldType (and item_type for lists) from observed values.

Checks bool before int deliberately — bool is an int subclass in Python, so the naive order mistypes every boolean column as INT.

Source code in graflo/architecture/onto_sample.py

def infer_field_type(values: list[Any]) -> tuple[FieldType, FieldType | None]:
    """Infer a ``FieldType`` (and ``item_type`` for lists) from observed values.

    Checks ``bool`` before ``int`` deliberately — ``bool`` is an ``int`` subclass
    in Python, so the naive order mistypes every boolean column as ``INT``.
    """
    non_null = [value for value in values if value is not None]
    if not non_null:
        return FieldType.STRING, None

    if all(isinstance(value, list) for value in non_null):
        items = [item for value in non_null for item in value]
        item_type, _ = infer_field_type(items) if items else (FieldType.STRING, None)
        return FieldType.LIST, item_type

    if all(isinstance(value, bool) for value in non_null):
        return FieldType.BOOL, None
    if all(
        isinstance(value, int) and not isinstance(value, bool) for value in non_null
    ):
        return FieldType.INT, None
    if all(
        isinstance(value, (int, float)) and not isinstance(value, bool)
        for value in non_null
    ):
        return FieldType.FLOAT, None
    if all(isinstance(value, (datetime, date, time)) for value in non_null):
        return FieldType.DATETIME, None

    if all(isinstance(value, str) for value in non_null):
        if all(UUID_PATTERN.match(value) for value in non_null):
            return FieldType.UUID, None
        if all(_ISO_DATETIME_PATTERN.match(value) for value in non_null):
            return FieldType.DATETIME, None

    return FieldType.STRING, None

iter_paths(doc, prefix='', depth=0)

Yield (path, depth, value) for every leaf in a JSON document.

Nested objects extend the path with .; lists of objects extend it with []. Lists of scalars are yielded whole so they can be typed as LIST.

Source code in graflo/architecture/onto_sample.py

def iter_paths(
    doc: dict[str, Any], prefix: str = "", depth: int = 0
) -> Iterator[tuple[str, int, Any]]:
    """Yield ``(path, depth, value)`` for every leaf in a JSON document.

    Nested objects extend the path with ``.``; lists of objects extend it with
    ``[]``. Lists of scalars are yielded whole so they can be typed as ``LIST``.
    """
    for key, value in doc.items():
        path = f"{prefix}.{key}" if prefix else key
        if isinstance(value, dict):
            yield from iter_paths(value, path, depth + 1)
        elif isinstance(value, list) and any(isinstance(item, dict) for item in value):
            nested_path = f"{path}{LIST_MARKER}"
            for item in value:
                if isinstance(item, dict):
                    yield from iter_paths(item, nested_path, depth + 1)
        else:
            yield path, depth, value

profile_sample(sample, *, max_paths=DEFAULT_MAX_PATHS, max_examples=DEFAULT_MAX_EXAMPLES)

Derive a :class:ResourceProfile from a sample's documents.

Handles tabular and hierarchical documents through one code path: a flat row is simply the depth-0 case.

Source code in graflo/architecture/onto_sample.py

def profile_sample(
    sample: ResourceSample,
    *,
    max_paths: int = DEFAULT_MAX_PATHS,
    max_examples: int = DEFAULT_MAX_EXAMPLES,
) -> ResourceProfile:
    """Derive a :class:`ResourceProfile` from a sample's documents.

    Handles tabular and hierarchical documents through one code path: a flat row
    is simply the depth-0 case.
    """
    observed: dict[str, list[Any]] = {}
    depths: dict[str, int] = {}

    for doc in sample.docs:
        for path, depth, value in iter_paths(doc):
            if path not in observed and len(observed) >= max_paths:
                continue
            observed.setdefault(path, []).append(value)
            depths[path] = max(depths.get(path, depth), depth)

    truncated = sample.truncated or len(observed) >= max_paths

    fields: list[FieldProfile] = []
    for path, values in observed.items():
        field_type, item_type = infer_field_type(values)
        non_null = [value for value in values if value is not None]
        hashable = {
            value if isinstance(value, (str, int, float, bool)) else repr(value)
            for value in non_null
        }
        fields.append(
            FieldProfile(
                path=path,
                type=field_type,
                item_type=item_type,
                depth=depths.get(path, 0),
                present=len(values),
                null_count=len(values) - len(non_null),
                distinct=len(hashable),
                examples=[_example(value) for value in non_null[:max_examples]],
                description=sample.field_descriptions.get(path),
            )
        )

    return ResourceProfile(
        resource_name=sample.resource_name,
        connector=sample.connector,
        description=sample.description,
        doc_count=len(sample.docs),
        max_depth=max(depths.values(), default=0),
        fields=fields,
        primary_key=list(sample.primary_key),
        foreign_keys=list(sample.foreign_keys),
        truncated=truncated,
    )

profile_source(source_sample, *, max_paths=DEFAULT_MAX_PATHS, max_examples=DEFAULT_MAX_EXAMPLES)

Profile every resource in a :class:SourceSample.

Source code in graflo/architecture/onto_sample.py

def profile_source(
    source_sample: SourceSample,
    *,
    max_paths: int = DEFAULT_MAX_PATHS,
    max_examples: int = DEFAULT_MAX_EXAMPLES,
) -> list[ResourceProfile]:
    """Profile every resource in a :class:`SourceSample`."""
    return [
        profile_sample(sample, max_paths=max_paths, max_examples=max_examples)
        for sample in source_sample.samples
    ]