Dictionary

cifflow.dictionary.ddlm_item

DDLm item definition — one save frame extracted from a DDLm dictionary.

DdlmItem dataclass

Represents a single definition extracted from a DDLm dictionary save frame.

Each save frame in a DDLm dictionary defines either a data item or a category. After import resolution, all relevant attributes are collected into this dataclass.

Attributes:

Name	Type	Description
definition_id	str	Canonical tag name as it appears in CIF data files, normalised to lowercase. Corresponds to _definition.id.
scope	str	"Item", "Category", or "Dictionary". Defaults to "Item" when _definition.scope is absent from the save frame.
definition_class	str	DDLm class of this definition: "Datum", "Attribute", "Loop", "Set", "Head", or "Functions". Defaults to "Datum" when _definition.class is absent.
category_id	str | None	SQLite table name derived from _name.category_id, lowercased. None for "Dictionary"-scope frames and items missing this tag.
object_id	str | None	SQLite column name derived from _name.object_id, lowercased. None for category frames and items missing this tag.
type_purpose	str | None	Value of _type.purpose (e.g. "Key", "Link", "SU", "Measurand"). None if absent.
type_source	str | None	Value of _type.source (e.g. "Assigned", "Recorded"). None if absent.
type_container	str	Value of _type.container (e.g. "Single", "List"). Defaults to "Single" when absent.
type_contents	str | None	Value of _type.contents (e.g. "Text", "Integer", "Real"). None if absent.
linked_item_id	str | None	For Link and SU items: the _definition.id of the linked item, lowercased. None for all other items.
units_code	str | None	Value of _units.code. None if absent.
description	str | None	Human-readable description from _description.text. None if absent.
enumeration_states	list[str]	Allowed enumeration values from _enumeration_set.state. Empty list when not present. Item-scope frames only.
enumeration_default	str | None	Default value from _enumeration.default. None if absent. The CIF inapplicable placeholder '.' is preserved as-is.
category_keys	list[str]	Lowercased fully-qualified tag names from _category_key.name. Empty list when not present. Category-scope frames only.
aliases	list[str]	Old tag names from _alias.definition_id, each mapping 1:1 to this definition_id. Empty list when none are declared.
replaced_by	list[str]	Preferred replacement tag names from _definition_replaced.by, lowercased. An empty string represents a PLACEHOLDER ("."), meaning deprecated with no replacement. Empty list when not present.
is_deprecated	bool	True if any _definition_replaced row exists for this item, regardless of the replacement value.
enumeration_range	str | None	Value of _enumeration.range. None if absent.
type_dimension	str | None	Value of _type.dimension. None if absent.
enumeration_def_index_ids	list[str]	Ordered list of canonical tag names from _enumeration.def_index_ids whose values form the lookup key for keyed defaults. Empty when absent.
enumeration_defaults	list[tuple[list[str], str]]	Keyed default table from a _enumeration_defaults loop: each entry is (key_components, default_value) where key_components aligns positionally with enumeration_def_index_ids. Empty when absent.
source_file	str | None	File path or URI of the dictionary file that defined this item. None when loaded from a JSON cache that predates this field.

Source code in src/cifflow/dictionary/ddlm_item.py

@dataclass
class DdlmItem:
    """
    Represents a single definition extracted from a DDLm dictionary save frame.

    Each save frame in a DDLm dictionary defines either a data item or a
    category.  After import resolution, all relevant attributes are collected
    into this dataclass.

    Attributes
    ----------
    definition_id:
        Canonical tag name as it appears in CIF data files, normalised to
        lowercase.  Corresponds to ``_definition.id``.
    scope:
        ``"Item"``, ``"Category"``, or ``"Dictionary"``.  Defaults to
        ``"Item"`` when ``_definition.scope`` is absent from the save frame.
    definition_class:
        DDLm class of this definition: ``"Datum"``, ``"Attribute"``,
        ``"Loop"``, ``"Set"``, ``"Head"``, or ``"Functions"``.  Defaults to
        ``"Datum"`` when ``_definition.class`` is absent.
    category_id:
        SQLite table name derived from ``_name.category_id``, lowercased.
        ``None`` for ``"Dictionary"``-scope frames and items missing this tag.
    object_id:
        SQLite column name derived from ``_name.object_id``, lowercased.
        ``None`` for category frames and items missing this tag.
    type_purpose:
        Value of ``_type.purpose`` (e.g. ``"Key"``, ``"Link"``, ``"SU"``,
        ``"Measurand"``).  ``None`` if absent.
    type_source:
        Value of ``_type.source`` (e.g. ``"Assigned"``, ``"Recorded"``).
        ``None`` if absent.
    type_container:
        Value of ``_type.container`` (e.g. ``"Single"``, ``"List"``).
        Defaults to ``"Single"`` when absent.
    type_contents:
        Value of ``_type.contents`` (e.g. ``"Text"``, ``"Integer"``,
        ``"Real"``).  ``None`` if absent.
    linked_item_id:
        For ``Link`` and ``SU`` items: the ``_definition.id`` of the linked
        item, lowercased.  ``None`` for all other items.
    units_code:
        Value of ``_units.code``.  ``None`` if absent.
    description:
        Human-readable description from ``_description.text``.  ``None`` if
        absent.
    enumeration_states:
        Allowed enumeration values from ``_enumeration_set.state``.  Empty
        list when not present.  Item-scope frames only.
    enumeration_default:
        Default value from ``_enumeration.default``.  ``None`` if absent.
        The CIF inapplicable placeholder ``'.'`` is preserved as-is.
    category_keys:
        Lowercased fully-qualified tag names from ``_category_key.name``.
        Empty list when not present.  Category-scope frames only.
    aliases:
        Old tag names from ``_alias.definition_id``, each mapping 1:1 to
        this ``definition_id``.  Empty list when none are declared.
    replaced_by:
        Preferred replacement tag names from ``_definition_replaced.by``,
        lowercased.  An empty string represents a ``PLACEHOLDER`` (``"."``),
        meaning deprecated with no replacement.  Empty list when not present.
    is_deprecated:
        ``True`` if any ``_definition_replaced`` row exists for this item,
        regardless of the replacement value.
    enumeration_range:
        Value of ``_enumeration.range``.  ``None`` if absent.
    type_dimension:
        Value of ``_type.dimension``.  ``None`` if absent.
    enumeration_def_index_ids:
        Ordered list of canonical tag names from ``_enumeration.def_index_ids``
        whose values form the lookup key for keyed defaults.  Empty when absent.
    enumeration_defaults:
        Keyed default table from a ``_enumeration_defaults`` loop: each entry is
        ``(key_components, default_value)`` where ``key_components`` aligns
        positionally with ``enumeration_def_index_ids``.  Empty when absent.
    source_file:
        File path or URI of the dictionary file that defined this item.
        ``None`` when loaded from a JSON cache that predates this field.
    """

    definition_id: str
    scope: str
    definition_class: str
    category_id: str | None
    object_id: str | None
    type_purpose: str | None
    type_source: str | None
    type_container: str
    type_contents: str | None
    linked_item_id: str | None
    units_code: str | None
    description: str | None
    enumeration_states: list[str] = field(default_factory=list)
    enumeration_default: str | None = None
    category_keys: list[str] = field(default_factory=list)
    aliases: list[str] = field(default_factory=list)
    replaced_by: list[str] = field(default_factory=list)
    is_deprecated: bool = False
    enumeration_range: str | None = None
    type_dimension: str | None = None
    enumeration_def_index_ids: list[str] = field(default_factory=list)
    enumeration_defaults: list[tuple[list[str], str]] = field(default_factory=list)
    source_file: str | None = None

cifflow.dictionary.ddlm_parser

DDLm dictionary data container produced by DictionaryLoader.

DdlmDictionary dataclass

In-memory representation of a loaded DDLm dictionary.

Produced by DictionaryLoader.load(). Contains all category and item definitions extracted from the dictionary's save frames, together with pre-built lookup tables for fast tag resolution.

Attributes:

Name	Type	Description
name	str	The data_ block name from the parsed CIF file (e.g. "CIF_CORE").
title	str | None	Value of _dictionary.title, or None if absent.
version	str | None	Value of _dictionary.version, or None if absent.
categories	dict[str, DdlmItem]	Mapping from lowercased definition_id to DdlmItem for every "Category"-scope frame.
items	dict[str, DdlmItem]	Mapping from lowercased definition_id to DdlmItem for every "Item"-scope frame.
tag_to_item	dict[str, DdlmItem]	Combined lookup covering every definition_id (both categories and items) plus all declared aliases. Keys are lowercased.
alias_to_definition_id	dict[str, str]	Maps each lowercased alias tag name to the current lowercased definition_id.
deprecated_ids	set[str]	Set of lowercased definition_id values whose definitions have been replaced (is_deprecated == True).
warnings	list[str]	Non-fatal issues encountered during loading, in emission order.
source_files	list[str]	Absolute file paths (or URIs) of every dictionary file loaded, in load order. Populated when path_resolver is supplied to :class:~cifflow.dictionary.loader.DictionaryLoader.
uri	str | None	Value of _dictionary.uri from the top-level dictionary source, or None if absent.

Source code in src/cifflow/dictionary/ddlm_parser.py

@dataclass
class DdlmDictionary:
    """
    In-memory representation of a loaded DDLm dictionary.

    Produced by ``DictionaryLoader.load()``.  Contains all category and item
    definitions extracted from the dictionary's save frames, together with
    pre-built lookup tables for fast tag resolution.

    Attributes
    ----------
    name:
        The ``data_`` block name from the parsed CIF file (e.g. ``"CIF_CORE"``).
    title:
        Value of ``_dictionary.title``, or ``None`` if absent.
    version:
        Value of ``_dictionary.version``, or ``None`` if absent.
    categories:
        Mapping from lowercased ``definition_id`` to ``DdlmItem`` for every
        ``"Category"``-scope frame.
    items:
        Mapping from lowercased ``definition_id`` to ``DdlmItem`` for every
        ``"Item"``-scope frame.
    tag_to_item:
        Combined lookup covering every ``definition_id`` (both categories and
        items) plus all declared aliases.  Keys are lowercased.
    alias_to_definition_id:
        Maps each lowercased alias tag name to the current lowercased
        ``definition_id``.
    deprecated_ids:
        Set of lowercased ``definition_id`` values whose definitions have been
        replaced (``is_deprecated == True``).
    warnings:
        Non-fatal issues encountered during loading, in emission order.
    source_files:
        Absolute file paths (or URIs) of every dictionary file loaded,
        in load order.  Populated when ``path_resolver`` is supplied to
        :class:`~cifflow.dictionary.loader.DictionaryLoader`.
    uri:
        Value of ``_dictionary.uri`` from the top-level dictionary source,
        or ``None`` if absent.
    """

    name: str
    title: str | None
    version: str | None
    categories: dict[str, DdlmItem]
    items: dict[str, DdlmItem]
    tag_to_item: dict[str, DdlmItem]
    alias_to_definition_id: dict[str, str]
    deprecated_ids: set[str]
    warnings: list[str] = field(default_factory=list)
    source_files: list[str] = field(default_factory=list)
    uri: str | None = None

    @property
    def has_warnings(self) -> bool:
        """Return True if any non-fatal issues were recorded during loading.

        Warnings range from purely informational (e.g. a dictionary with
        multiple data blocks) to significant (e.g. a skipped import).  A
        dictionary with warnings is not necessarily unusable; inspect
        :attr:`warnings` to determine severity.
        """
        return bool(self.warnings)

has_warnings property

Return True if any non-fatal issues were recorded during loading.

Warnings range from purely informational (e.g. a dictionary with multiple data blocks) to significant (e.g. a skipped import). A dictionary with warnings is not necessarily unusable; inspect :attr:warnings to determine severity.

cifflow.dictionary.loader

DDLm dictionary loader — parses a DDLm CIF and resolves _import.get directives.

SourceResolver = Callable[[str], str | None] module-attribute

Callable that maps a URI string to a raw CIF source string, or None.

DictionaryLoader

Loads a DDLm dictionary from a CIF 2.0 source string.

Resolves _import.get directives using the supplied SourceResolver. Both mode="Contents" (frame-level attribute merge) and mode="Full" (constituent dictionary incorporation) are supported. File access is fully delegated to the resolver; this class never accesses the filesystem or network directly.

Parsed files are cached for the lifetime of the loader instance. To invalidate the cache, create a new instance.

Parameters:

Name	Type	Description	Default
resolver	SourceResolver | None	Callable that maps a URI string to a raw CIF source string, or None if the file is unavailable. If None, import directives that require an external file will trigger the if_miss policy.	None
path_resolver	Callable[[str], str | None] | None	Optional companion to resolver that maps the same URI to an absolute filesystem path. When provided, the resolved paths are recorded in :attr:~cifflow.dictionary.ddlm_parser.DdlmDictionary.source_files.	None
on_warning	Callable[[str], None] | None	Optional callback for non-fatal warnings. If None, warnings are silently discarded.	None
ignore_head_imports	bool	When True, _import.get directives in save frames with _definition.class = Head are silently skipped. Only the save frames physically present in the file being loaded are parsed. Applies to all files loaded by this instance, including constituents loaded via mode="Full" recursion. Defaults to False.	False
block_constituent_imports	bool	When True, mode="Full" imports whose target save frame has _definition.class = Head (whole-dictionary constituent imports) are skipped with a warning. Frame-level attribute imports (mode="Contents" or mode="Full" targeting a non-Head frame) are unaffected. Defaults to False.	False

Source code in src/cifflow/dictionary/loader.py

class DictionaryLoader:
    """
    Loads a DDLm dictionary from a CIF 2.0 source string.

    Resolves ``_import.get`` directives using the supplied ``SourceResolver``.
    Both ``mode="Contents"`` (frame-level attribute merge) and ``mode="Full"``
    (constituent dictionary incorporation) are supported.  File access is fully
    delegated to the resolver; this class never accesses the filesystem or
    network directly.

    Parsed files are cached for the lifetime of the loader instance.  To
    invalidate the cache, create a new instance.

    Parameters
    ----------
    resolver
        Callable that maps a URI string to a raw CIF source string, or ``None``
        if the file is unavailable.  If ``None``, import directives that require
        an external file will trigger the ``if_miss`` policy.
    path_resolver
        Optional companion to *resolver* that maps the same URI to an absolute
        filesystem path.  When provided, the resolved paths are recorded in
        :attr:`~cifflow.dictionary.ddlm_parser.DdlmDictionary.source_files`.
    on_warning
        Optional callback for non-fatal warnings.  If ``None``, warnings are
        silently discarded.
    ignore_head_imports
        When ``True``, ``_import.get`` directives in save frames with
        ``_definition.class = Head`` are silently skipped.  Only the save
        frames physically present in the file being loaded are parsed.
        Applies to all files loaded by this instance, including constituents
        loaded via ``mode="Full"`` recursion.  Defaults to ``False``.
    block_constituent_imports
        When ``True``, ``mode="Full"`` imports whose target save frame has
        ``_definition.class = Head`` (whole-dictionary constituent imports) are
        skipped with a warning.  Frame-level attribute imports (``mode="Contents"``
        or ``mode="Full"`` targeting a non-Head frame) are unaffected.
        Defaults to ``False``.
    """

    def __init__(
        self,
        resolver: SourceResolver | None = None,
        *,
        path_resolver: 'Callable[[str], str | None] | None' = None,
        on_warning: Callable[[str], None] | None = None,
        ignore_head_imports: bool = False,
        block_constituent_imports: bool = False,
    ) -> None:
        self._resolver = resolver
        self._path_resolver = path_resolver
        self._on_warning = on_warning if on_warning is not None else lambda msg: None
        self._ignore_head_imports = ignore_head_imports
        self._block_constituent_imports = block_constituent_imports
        self._source_cache: dict[str, str] = {}
        self._parse_cache: dict[str, CifFile] = {}

    def load(self, source: str, *, base_uri: str | None = None) -> DdlmDictionary:
        """
        Parse a DDLm dictionary source string and resolve all ``_import.get`` directives.

        Both ``mode="Contents"`` (frame-level attribute merge) and
        ``mode="Full"`` (constituent dictionary incorporation) are supported.
        When a ``mode="Full"`` import targets a Head category, the entire
        constituent dictionary is loaded recursively and its definitions are
        merged into the result, with local definitions taking precedence.

        Circular imports are detected and skipped with a warning.

        Parameters
        ----------
        source:
            Raw CIF 2.0 source string of the dictionary to parse.
        base_uri:
            URI of the dictionary being parsed, used as the base for resolving
            relative import URIs.  If ``None`` and ``_dictionary.uri`` is
            present in the dictionary, that value is used.  If neither is
            available, relative URIs are passed to the resolver as-is.

        Returns
        -------
        DdlmDictionary
            The fully loaded dictionary with all imports resolved.
        """
        collected: list[str] = []
        if base_uri:
            resolved = self._path_resolver(base_uri) if self._path_resolver else None
            collected.append(resolved or base_uri)
        return self._load_recursive(source, base_uri, set(), collected)

    def _load_recursive(
        self,
        source: str,
        base_uri: str | None,
        loading: set[str],
        collected: list[str],
    ) -> DdlmDictionary:
        """Parse and resolve one dictionary, tracking *loading* for cycle detection."""
        warnings: list[str] = []

        def warn(msg: str) -> None:
            warnings.append(msg)
            self._on_warning(msg)

        cif, parse_errors = build(source)
        for e in parse_errors:
            warn(f'parse error in dictionary: {e.message} (line {e.line})')

        if not cif.blocks:
            warn('dictionary CIF contains no data blocks')
            return DdlmDictionary(
                name='', title=None, version=None,
                categories={}, items={}, tag_to_item={},
                alias_to_definition_id={}, deprecated_ids=set(),
                warnings=warnings,
            )

        if len(cif.blocks) > 1:
            warn(f'dictionary CIF has {len(cif.blocks)} data blocks — using first')

        block_name = cif.blocks[0]
        block = cif[block_name]

        # Read the canonical dictionary URI unconditionally.
        uri_vals = block['_dictionary.uri'] if '_dictionary.uri' in block else []
        dict_uri = uri_vals[0] if uri_vals and isinstance(uri_vals[0], str) and uri_vals[0] not in ('.', '?') else None

        # Resolve base_uri for import resolution if not supplied by caller.
        if base_uri is None:
            base_uri = dict_uri

        title = block['_dictionary.title'][0] if '_dictionary.title' in block else None
        if isinstance(title, str) and title in ('.', '?'):
            title = None
        version = block['_dictionary.version'][0] if '_dictionary.version' in block else None
        if isinstance(version, str) and version in ('.', '?'):
            version = None

        # pool accumulates DdlmItems from mode="Full" constituent imports.
        # Primary items (from this file's frames) are appended afterwards so
        # they overwrite constituent definitions with the same definition_id.
        pool: dict[str, DdlmItem] = {}
        primary_items: list[DdlmItem] = []

        for sf_name in block.save_frames:
            sf = block[sf_name]
            frame_data = {tag: sf[tag] for tag in sf.tags if tag in _FRAME_TAGS}
            _apply_tag_aliases(frame_data)

            frame_class = (_scalar(frame_data, '_definition.class') or '').lower()
            is_head = frame_class == 'head'
            if '_import.get' in frame_data and not (self._ignore_head_imports and is_head):
                directives_val = frame_data['_import.get']
                if directives_val and isinstance(directives_val[0], list):
                    directives = directives_val[0]
                    self._resolve_imports(
                        frame_data, directives, base_uri, loading, pool, warn, collected
                    )

            item = _extract_item(frame_data, warn, source_file=_current_source(base_uri, self._path_resolver))
            if item is not None:
                primary_items.append(item)

        # Merge: constituents first (pool), then primary overwrites.
        # Deduplicate by definition_id so _build_lookup_tables only sees one
        # item per id and doesn't generate spurious alias collision warnings.
        _merged: dict[str, DdlmItem] = {item.definition_id: item for item in pool.values()}
        for item in primary_items:
            _merged[item.definition_id] = item
        all_items = list(_merged.values())

        categories, items, tag_to_item, alias_to_def_id, deprecated_ids = (
            _build_lookup_tables(all_items, warn)
        )

        return DdlmDictionary(
            name=block_name,
            title=title,
            version=version,
            uri=dict_uri,
            categories=categories,
            items=items,
            tag_to_item=tag_to_item,
            alias_to_definition_id=alias_to_def_id,
            deprecated_ids=deprecated_ids,
            warnings=warnings,
            source_files=list(collected),
        )

    def _load_constituent(
        self,
        uri: str,
        loading: set[str],
        warn: Callable[[str], None],
        collected: list[str] | None = None,
    ) -> DdlmDictionary | None:
        """
        Load and return the dictionary at *uri*, or ``None`` on failure.

        Checks *loading* for circular imports before proceeding.  Adds *uri*
        to *loading* for the duration of the recursive call.
        """
        if uri in loading:
            warn(f'circular import detected for {uri!r} — skipped')
            return None
        src = self._get_source(uri)
        if src is None:
            return None
        if collected is not None:
            resolved = self._path_resolver(uri) if self._path_resolver else None
            entry = resolved or uri
            if entry not in collected:
                collected.append(entry)
        loading.add(uri)
        try:
            return self._load_recursive(src, uri, loading, collected if collected is not None else [])
        finally:
            loading.discard(uri)

    def _resolve_imports(
        self,
        frame_data: dict[str, list],
        directives: list[Any],
        base_uri: str | None,
        loading: set[str],
        pool: dict[str, DdlmItem],
        warn: Callable[[str], None],
        collected: list[str] | None = None,
    ) -> None:
        """Apply ``_import.get`` directives to *frame_data* and/or *pool*."""
        # Sort by 'order' if present; fall back to list order.
        def _order_key(d: Any) -> int:
            if not isinstance(d, dict):
                return 0
            v = d.get('order')
            try:
                return int(v) if v is not None else 0
            except (TypeError, ValueError):
                return 0

        sorted_directives = sorted(directives, key=_order_key)

        for directive in sorted_directives:
            if not isinstance(directive, dict):
                warn(f'_import.get directive is not a table — skipped: {directive!r}')
                continue

            file_uri = directive.get('file', '')
            save_id = directive.get('save', '')
            mode = directive.get('mode', 'Contents')
            dupl = directive.get('dupl', 'Exit')
            miss = directive.get('miss', 'Exit')

            if not isinstance(file_uri, str) or not file_uri:
                warn("_import.get directive missing 'file' key — skipped")
                continue
            if not isinstance(save_id, str) or not save_id:
                warn("_import.get directive missing 'save' key — skipped")
                continue

            if mode not in ('Contents', 'Full'):
                warn(
                    f"_import.get mode {mode!r} is not supported "
                    f"(file={file_uri!r}, save={save_id!r}) — skipped"
                )
                continue

            # Resolve the URI relative to base_uri if needed.
            resolved_uri = self._resolve_uri(file_uri, base_uri)

            if mode == 'Full':
                # Look up the named save frame first to determine whether the
                # target is a Head category (dictionary-level import) or an
                # ordinary frame (frame-level attribute merge like Contents).
                source_cif = self._get_parsed(resolved_uri)

                if source_cif is None:
                    msg = (
                        f"_import.get could not load {resolved_uri!r} "
                        f"(save={save_id!r})"
                    )
                    if miss == 'Ignore':
                        warn(msg + ' — ignored')
                        continue
                    else:
                        warn(msg + ' — aborting dictionary load')
                        return

                source_frame_data = self._find_frame_by_definition_id(
                    source_cif, save_id, lambda _: None
                )

                if source_frame_data is None:
                    msg = (
                        f"_import.get save frame {save_id!r} not found "
                        f"in {resolved_uri!r}"
                    )
                    if miss == 'Ignore':
                        warn(msg + ' — ignored')
                        continue
                    else:
                        warn(msg + ' — aborting dictionary load')
                        return

                target_class = (
                    _scalar(source_frame_data, '_definition.class') or ''
                ).lower()

                if target_class == 'head':
                    if self._block_constituent_imports:
                        warn(
                            f"_import.get constituent dictionary import blocked "
                            f"(file={resolved_uri!r}, save={save_id!r})"
                        )
                        continue
                    # Dictionary-level import: load the entire constituent
                    # dictionary and merge all its definitions into pool.
                    constituent = self._load_constituent(resolved_uri, loading, warn, collected)
                    if constituent is None:
                        msg = (
                            f"_import.get could not load constituent "
                            f"{resolved_uri!r} (save={save_id!r})"
                        )
                        if miss == 'Ignore':
                            warn(msg + ' — ignored')
                            continue
                        else:
                            warn(msg + ' — aborting dictionary load')
                            return

                    # Surface constituent warnings prefixed with their source.
                    for w in constituent.warnings:
                        warn(f'[{resolved_uri}] {w}')

                    abort = _merge_constituent(pool, constituent, dupl, warn)
                    if abort:
                        return
                    continue

                # Non-Head target: frame-level attribute merge (same as Contents).
                # Fall through to the shared frame-merge path below.
                # source_cif and source_frame_data are already resolved above.

            else:
                # mode == 'Contents': frame-level attribute merge.
                source_cif = self._get_parsed(resolved_uri)

                if source_cif is None:
                    msg = (
                        f"_import.get could not load {resolved_uri!r} "
                        f"(save={save_id!r})"
                    )
                    if miss == 'Ignore':
                        warn(msg + ' — ignored')
                        continue
                    else:
                        warn(msg + ' — aborting dictionary load')
                        return

                source_frame_data = self._find_frame_by_definition_id(
                    source_cif, save_id, warn
                )

                if source_frame_data is None:
                    msg = (
                        f"_import.get save frame with _definition.id={save_id!r} "
                        f"not found in {resolved_uri!r}"
                    )
                    if miss == 'Ignore':
                        warn(msg + ' — ignored')
                        continue
                    else:
                        warn(msg + ' — aborting dictionary load')
                        return

            # Shared frame-level merge path (mode="Contents" or mode="Full" non-Head).
            # source_cif and source_frame_data are already resolved above.
            abort = self._merge_frame(
                frame_data, source_frame_data, source_cif, dupl, warn
            )
            if abort:
                return

    def _resolve_uri(self, uri: str, base_uri: str | None) -> str:
        """Return the URI to pass to the resolver."""
        # If the URI looks absolute or base_uri is absent, use it as-is.
        return uri

    def _get_source(self, uri: str) -> str | None:
        """Return raw CIF source for *uri*, using cache then resolver."""
        if uri in self._source_cache:
            return self._source_cache[uri]
        if self._resolver is None:
            return None
        src = self._resolver(uri)
        if src is not None:
            self._source_cache[uri] = src
        return src

    def _get_parsed(self, uri: str) -> CifFile | None:
        """Return a parsed CifFile for *uri*, using cache then resolver."""
        if uri in self._parse_cache:
            return self._parse_cache[uri]
        src = self._get_source(uri)
        if src is None:
            return None
        cif, _ = build(src)
        self._parse_cache[uri] = cif
        return cif

    def _find_frame_by_definition_id(
        self,
        cif: CifFile,
        definition_id: str,
        warn: Callable[[str], None],
    ) -> dict[str, list] | None:
        """
        Search all save frames in *cif* for one matching *definition_id*.

        Match strategy (case-insensitive, in priority order):

        1. ``_definition.id`` value — used by full dictionary frames.
        2. Save frame label — used by template files (e.g. ``templ_attr.cif``)
           that declare no ``_definition.id``.

        Returns the frame's working dict filtered to ``_FRAME_TAGS``, or
        ``None`` if no match is found.
        """
        if not cif.blocks:
            return None
        block = cif[cif.blocks[0]]
        target = definition_id.lower()
        for sf_name in block.save_frames:
            sf = block[sf_name]
            if '_definition.id' in sf:
                raw_id = sf['_definition.id'][0]
                if isinstance(raw_id, str) and raw_id.lower() == target:
                    fd = {tag: sf[tag] for tag in sf.tags if tag in _FRAME_TAGS}
                    _apply_tag_aliases(fd)
                    return fd
            elif sf_name.lower() == target:
                # Template files carry no _definition.id; match by frame label.
                fd = {tag: sf[tag] for tag in sf.tags if tag in _FRAME_TAGS}
                _apply_tag_aliases(fd)
                return fd
        return None

    def _merge_frame(
        self,
        frame_data: dict[str, list],
        source_data: dict[str, list],
        source_cif: CifFile,
        dupl: str,
        warn: Callable[[str], None],
    ) -> bool:
        """
        Merge *source_data* tags into *frame_data* according to *dupl* policy.

        Returns ``True`` if the load should be aborted (``dupl == "Exit"``
        and a conflict was found), ``False`` otherwise.
        """
        for tag, values in source_data.items():
            if tag in _IMPORT_IDENTITY_TAGS:
                # Never import frame-identity tags from a source frame.
                continue
            if tag not in frame_data:
                frame_data[tag] = values
            else:
                if dupl == 'Ignore':
                    pass  # Keep existing value.
                elif dupl == 'Replace':
                    # If the tag belongs to a Loop category, remove all tags
                    # from that category in frame_data before inserting.
                    self._replace_loop_category_tags(
                        frame_data, tag, source_cif
                    )
                    frame_data[tag] = values
                else:  # 'Exit' (default)
                    warn(
                        f"_import.get dupl=Exit: conflict on tag {tag!r} — "
                        f"aborting dictionary load"
                    )
                    return True
        return False

    def _replace_loop_category_tags(
        self,
        frame_data: dict[str, list],
        tag: str,
        source_cif: CifFile,
    ) -> None:
        """If *tag* belongs to a Loop category in *source_cif*, remove all tags from that category in *frame_data* before the caller inserts the new value."""
        if not source_cif.blocks:
            return
        block = source_cif[source_cif.blocks[0]]

        # Find the tag's save frame to get its _name.category_id.
        tag_lower = tag.lower()
        category_id: str | None = None
        for sf_name in block.save_frames:
            sf = block[sf_name]
            if '_definition.id' not in sf:
                continue
            raw_id = sf['_definition.id'][0]
            if not isinstance(raw_id, str) or raw_id.lower() != tag_lower:
                continue
            cat_vals = sf['_name.category_id'] if '_name.category_id' in sf else []
            if cat_vals and isinstance(cat_vals[0], str):
                category_id = cat_vals[0].lower()
            break

        if category_id is None:
            return

        # Check if that category is a Loop class.
        for sf_name in block.save_frames:
            sf = block[sf_name]
            if '_definition.id' not in sf:
                continue
            raw_id = sf['_definition.id'][0]
            if not isinstance(raw_id, str) or raw_id.lower() != category_id:
                continue
            class_vals = sf['_definition.class'] if '_definition.class' in sf else []
            if class_vals and isinstance(class_vals[0], str):
                if class_vals[0].lower() == 'loop':
                    # Remove all tags in frame_data that belong to this category.
                    # Look up each tag's category via its save frame in source_cif.
                    self._remove_category_tags(frame_data, category_id, block)
            break

    def _remove_category_tags(
        self,
        frame_data: dict[str, list],
        category_id: str,
        block: Any,
    ) -> None:
        """Remove all tags from *frame_data* whose category is *category_id*."""
        to_remove = []
        for existing_tag in list(frame_data):
            for sf_name in block.save_frames:
                sf = block[sf_name]
                if '_definition.id' not in sf:
                    continue
                raw_id = sf['_definition.id'][0]
                if not isinstance(raw_id, str) or raw_id.lower() != existing_tag.lower():
                    continue
                cat_vals = sf['_name.category_id'] if '_name.category_id' in sf else []
                if cat_vals and isinstance(cat_vals[0], str):
                    if cat_vals[0].lower() == category_id:
                        to_remove.append(existing_tag)
                break
        for t in to_remove:
            del frame_data[t]

load(source, *, base_uri=None)

Parse a DDLm dictionary source string and resolve all _import.get directives.

Both mode="Contents" (frame-level attribute merge) and mode="Full" (constituent dictionary incorporation) are supported. When a mode="Full" import targets a Head category, the entire constituent dictionary is loaded recursively and its definitions are merged into the result, with local definitions taking precedence.

Circular imports are detected and skipped with a warning.

Parameters:

Name	Type	Description	Default
source	str	Raw CIF 2.0 source string of the dictionary to parse.	required
base_uri	str | None	URI of the dictionary being parsed, used as the base for resolving relative import URIs. If None and _dictionary.uri is present in the dictionary, that value is used. If neither is available, relative URIs are passed to the resolver as-is.	None

Returns:

Type	Description
DdlmDictionary	The fully loaded dictionary with all imports resolved.

Source code in src/cifflow/dictionary/loader.py

def load(self, source: str, *, base_uri: str | None = None) -> DdlmDictionary:
    """
    Parse a DDLm dictionary source string and resolve all ``_import.get`` directives.

    Both ``mode="Contents"`` (frame-level attribute merge) and
    ``mode="Full"`` (constituent dictionary incorporation) are supported.
    When a ``mode="Full"`` import targets a Head category, the entire
    constituent dictionary is loaded recursively and its definitions are
    merged into the result, with local definitions taking precedence.

    Circular imports are detected and skipped with a warning.

    Parameters
    ----------
    source:
        Raw CIF 2.0 source string of the dictionary to parse.
    base_uri:
        URI of the dictionary being parsed, used as the base for resolving
        relative import URIs.  If ``None`` and ``_dictionary.uri`` is
        present in the dictionary, that value is used.  If neither is
        available, relative URIs are passed to the resolver as-is.

    Returns
    -------
    DdlmDictionary
        The fully loaded dictionary with all imports resolved.
    """
    collected: list[str] = []
    if base_uri:
        resolved = self._path_resolver(base_uri) if self._path_resolver else None
        collected.append(resolved or base_uri)
    return self._load_recursive(source, base_uri, set(), collected)

directory_resolver(path)

Return a SourceResolver that reads files by filename from a local directory.

The last path component of the URI is used as the filename. Returns None if the file is not found in the directory.

Parameters:

Name	Type	Description	Default
path	str | Path	Directory to search for dictionary files.	required

Returns:

Type	Description
SourceResolver	A callable mapping URI strings to raw CIF source strings.

Source code in src/cifflow/dictionary/loader.py

def directory_resolver(path: str | pathlib.Path) -> SourceResolver:
    """
    Return a SourceResolver that reads files by filename from a local directory.

    The last path component of the URI is used as the filename.  Returns
    ``None`` if the file is not found in the directory.

    Parameters
    ----------
    path:
        Directory to search for dictionary files.

    Returns
    -------
    SourceResolver
        A callable mapping URI strings to raw CIF source strings.
    """
    directory = pathlib.Path(path)

    def _resolve(uri: str) -> str | None:
        filename = pathlib.PurePosixPath(uri).name
        candidate = directory / filename
        if candidate.exists():
            return candidate.read_text(encoding='utf-8')
        return None

    return _resolve

directory_path_resolver(path)

Return a path resolver that maps a URI to its absolute file path.

Companion to :func:directory_resolver. Pass to DictionaryLoader(path_resolver=...) so that source_files in the resulting DdlmDictionary contains absolute paths rather than bare URIs.

Parameters:

Name	Type	Description	Default
path	str | Path	Directory to search for dictionary files.	required

Returns:

Type	Description
Callable[[str], str | None]	Maps URI strings to absolute path strings, or None if not found.

Source code in src/cifflow/dictionary/loader.py

def directory_path_resolver(path: str | pathlib.Path) -> 'Callable[[str], str | None]':
    """
    Return a path resolver that maps a URI to its absolute file path.

    Companion to :func:`directory_resolver`.  Pass to
    ``DictionaryLoader(path_resolver=...)`` so that ``source_files`` in the
    resulting ``DdlmDictionary`` contains absolute paths rather than bare URIs.

    Parameters
    ----------
    path:
        Directory to search for dictionary files.

    Returns
    -------
    Callable[[str], str | None]
        Maps URI strings to absolute path strings, or ``None`` if not found.
    """
    directory = pathlib.Path(path)

    def _resolve_path(uri: str) -> str | None:
        filename = pathlib.PurePosixPath(uri).name
        candidate = directory / filename
        if candidate.exists():
            return str(candidate.resolve())
        return None

    return _resolve_path

merge_dictionaries(*dictionaries, dupl='Ignore', on_warning=lambda _: None)

Merge two or more loaded DdlmDictionary objects into one.

Dictionaries are processed left-to-right. The dupl policy controls what happens when the same definition_id appears in more than one input:

• 'Ignore' (default): the first definition seen is kept.
• 'Replace': the later dictionary's definition wins.
• 'Exit': raise ValueError if any duplicate definition is found.

The merged dictionary's name, title, version, and uri are taken from the first input. source_files and warnings are the concatenation of all inputs.

Parameters:

Name	Type	Description	Default
*dictionaries	DdlmDictionary	Two or more DdlmDictionary instances to merge.	()
dupl	str	Conflict resolution policy: 'Ignore', 'Replace', or 'Exit'.	'Ignore'
on_warning	Callable[[str], None]	Called with each warning message (e.g. print).	lambda _: None

Returns:

Type	Description
DdlmDictionary	A new merged dictionary.

Raises:

Type	Description
ValueError	If fewer than two dictionaries are supplied, dupl is invalid, or dupl is 'Exit' and a duplicate definition is encountered.

Source code in src/cifflow/dictionary/loader.py

def merge_dictionaries(
    *dictionaries: DdlmDictionary,
    dupl: str = 'Ignore',
    on_warning: Callable[[str], None] = lambda _: None,
) -> DdlmDictionary:
    """Merge two or more loaded ``DdlmDictionary`` objects into one.

    Dictionaries are processed left-to-right.  The *dupl* policy controls what
    happens when the same ``definition_id`` appears in more than one input:

    - ``'Ignore'`` *(default)*: the first definition seen is kept.
    - ``'Replace'``: the later dictionary's definition wins.
    - ``'Exit'``: raise ``ValueError`` if any duplicate definition is found.

    The merged dictionary's ``name``, ``title``, ``version``, and ``uri`` are
    taken from the first input.  ``source_files`` and ``warnings`` are the
    concatenation of all inputs.

    Parameters
    ----------
    *dictionaries:
        Two or more ``DdlmDictionary`` instances to merge.
    dupl:
        Conflict resolution policy: ``'Ignore'``, ``'Replace'``, or ``'Exit'``.
    on_warning:
        Called with each warning message (e.g. ``print``).

    Returns
    -------
    DdlmDictionary
        A new merged dictionary.

    Raises
    ------
    ValueError
        If fewer than two dictionaries are supplied, *dupl* is invalid, or
        *dupl* is ``'Exit'`` and a duplicate definition is encountered.
    """
    if len(dictionaries) < 2:
        raise ValueError('merge_dictionaries requires at least two dictionaries')
    if dupl not in ('Replace', 'Ignore', 'Exit'):
        raise ValueError(f"dupl must be 'Replace', 'Ignore', or 'Exit', got {dupl!r}")

    warnings: list[str] = []

    def warn(msg: str) -> None:
        warnings.append(msg)
        on_warning(msg)

    pool: dict[str, DdlmItem] = {}
    source_files: list[str] = []

    for d in dictionaries:
        warnings.extend(d.warnings)
        source_files.extend(d.source_files)
        if dupl == 'Exit':
            conflicts = [
                def_id for def_id in {**d.categories, **d.items}
                if def_id in pool
            ]
            if conflicts:
                raise ValueError(
                    f'duplicate definition(s) found while merging dictionaries: '
                    + ', '.join(sorted(conflicts))
                )
        _merge_constituent(pool, d, dupl, warn)

    categories, items, tag_to_item, alias_to_def_id, deprecated_ids = (
        _build_lookup_tables(list(pool.values()), warn)
    )

    first = dictionaries[0]
    return DdlmDictionary(
        name=first.name,
        title=first.title,
        version=first.version,
        uri=first.uri,
        categories=categories,
        items=items,
        tag_to_item=tag_to_item,
        alias_to_definition_id=alias_to_def_id,
        deprecated_ids=deprecated_ids,
        warnings=warnings,
        source_files=source_files,
    )

cifflow.dictionary.schema

SQLite schema generation from a loaded DDLm dictionary.

BridgeColumnDef dataclass

A column whose value is derived transitively through one or more tables.

When populating table_name, the column column_name has no direct CIF source. Its value is resolved by following a chain of single-column FK lookups described by hops, then reading bridge_value_column from the final table in the chain.

Each hop is a 3-tuple (via_column, bridge_table, bridge_pk_column):

• via_column: column in the previous table (or in table_name for the first hop) whose value is used as the lookup key.
• bridge_table: the table to look up in.
• bridge_pk_column: the PK column of bridge_table matched against via_column.

For a single-hop bridge the chain has length 1 and the semantics are identical to the legacy four-field form.

Attributes:

Name	Type	Description
table_name	str	Table that gains the derived column (e.g. 'geom_angle').
column_name	str	Name of the derived column (e.g. 'structure_id').
hops	list[tuple[str, str, str]]	Ordered list of (via_column, bridge_table, bridge_pk_column) tuples, one per lookup step. Must contain at least one entry.
bridge_value_column	str	Column in the last hop's bridge_table whose value is copied into column_name (e.g. 'structure_id').
fallback_chains	list[tuple[list[tuple[str, str, str]], str]]	Alternative resolution chains tried in order when the primary chain yields None for a given row. Each entry is a (hops, bridge_value_column) pair with the same structure as the primary fields.

Source code in src/cifflow/dictionary/schema.py

@dataclass
class BridgeColumnDef:
    """
    A column whose value is derived transitively through one or more tables.

    When populating ``table_name``, the column ``column_name`` has no direct
    CIF source.  Its value is resolved by following a chain of single-column
    FK lookups described by ``hops``, then reading ``bridge_value_column``
    from the final table in the chain.

    Each hop is a 3-tuple ``(via_column, bridge_table, bridge_pk_column)``:

    - ``via_column``: column in the *previous* table (or in ``table_name``
      for the first hop) whose value is used as the lookup key.
    - ``bridge_table``: the table to look up in.
    - ``bridge_pk_column``: the PK column of ``bridge_table`` matched
      against ``via_column``.

    For a single-hop bridge the chain has length 1 and the semantics are
    identical to the legacy four-field form.

    Attributes
    ----------
    table_name:
        Table that gains the derived column (e.g. ``'geom_angle'``).
    column_name:
        Name of the derived column (e.g. ``'structure_id'``).
    hops:
        Ordered list of ``(via_column, bridge_table, bridge_pk_column)``
        tuples, one per lookup step.  Must contain at least one entry.
    bridge_value_column:
        Column in the *last* hop's ``bridge_table`` whose value is copied
        into ``column_name`` (e.g. ``'structure_id'``).
    fallback_chains:
        Alternative resolution chains tried in order when the primary chain
        yields ``None`` for a given row.  Each entry is a
        ``(hops, bridge_value_column)`` pair with the same structure as the
        primary fields.
    """

    table_name: str
    column_name: str
    hops: list[tuple[str, str, str]]
    bridge_value_column: str
    fallback_chains: 'list[tuple[list[tuple[str, str, str]], str]]' = field(default_factory=list)

    # ------------------------------------------------------------------
    # Backward-compat properties (single-hop case; also useful for
    # visualisation which only needs the first and last table).
    # ------------------------------------------------------------------

    @property
    def via_column(self) -> str:
        """Via-column of the first hop (column in ``table_name``)."""
        return self.hops[0][0]

    @property
    def bridge_table(self) -> str:
        """Bridge table of the last hop (the table holding the value)."""
        return self.hops[-1][1]

    @property
    def bridge_pk_column(self) -> str:
        """PK column of the last hop's bridge table."""
        return self.hops[-1][2]

via_column property

Via-column of the first hop (column in table_name).

bridge_table property

Bridge table of the last hop (the table holding the value).

bridge_pk_column property

PK column of the last hop's bridge table.

ForeignKeyDef dataclass

A FOREIGN KEY constraint between two tables (single- or multi-column).

Always emitted with DEFERRABLE INITIALLY DEFERRED to handle cyclic category graphs correctly within a transaction.

Attributes:

Name	Type	Description
source_table	str	Name of the table that holds the foreign key column(s).
source_columns	list[str]	Ordered list of foreign key column names in source_table.
target_table	str	Name of the table being referenced.
target_columns	list[str]	Ordered list of column names being referenced in target_table, corresponding positionally to source_columns.

Source code in src/cifflow/dictionary/schema.py

@dataclass
class ForeignKeyDef:
    """
    A ``FOREIGN KEY`` constraint between two tables (single- or multi-column).

    Always emitted with ``DEFERRABLE INITIALLY DEFERRED`` to handle cyclic
    category graphs correctly within a transaction.

    Attributes
    ----------
    source_table:
        Name of the table that holds the foreign key column(s).
    source_columns:
        Ordered list of foreign key column names in *source_table*.
    target_table:
        Name of the table being referenced.
    target_columns:
        Ordered list of column names being referenced in *target_table*,
        corresponding positionally to *source_columns*.
    """

    source_table: str
    source_columns: list[str]
    target_table: str
    target_columns: list[str]

ColumnDef dataclass

Definition of a single column in a generated SQLite table.

Attributes:

Name	Type	Description
name	str	SQL column name, equal to the DDLm _name.object_id, lowercased. For synthetic columns the name is _cifflow_block_id, _cifflow_row_id, or _cifflow_id.
definition_id	str	The current canonical _definition.id for this column's DDLm item. Empty string for synthetic columns.
type_contents	str | None	DDLm _type.contents value (e.g. "Text", "Integer", "Real", "List"); None if absent from the dictionary or for synthetic columns. Informational only -- DDL always emits TEXT for all value columns; _cifflow_row_id always emits INTEGER.
nullable	bool	False for synthetic and primary-key columns; True for all other domain columns.
is_primary_key	bool	True if this column is part of the table's PRIMARY KEY.
is_synthetic	bool	True for the _cifflow_block_id, _cifflow_row_id, and _cifflow_id infrastructure columns, which have no corresponding DDLm item definition.
linked_item_id	str | None	For SU items only: the _definition.id of the associated measurand item, lowercased. None for all other column types. Does not produce a FOREIGN KEY constraint; used by the ingestion and output layers.
type_container	str | None	DDLm _type.container value (e.g. "Single", "List", "Matrix"); None for synthetic columns, "Single" as the DDLm default for domain columns when the attribute is absent. Non-"Single" containers store JSON text in SQLite regardless of type_contents.
enumeration_states	list[str]	Allowed enumeration values from _enumeration_set.state. Empty list when not present.
enumeration_range	str | None	Value of _enumeration.range. None if absent.
type_dimension	str | None	Value of _type.dimension. None if absent.
enumeration_default	str | None	Scalar default value from _enumeration.default. None if absent.
enumeration_def_index_ids	list[str]	Ordered index tag names from _enumeration.def_index_ids for keyed default lookup. Empty when absent.
enumeration_defaults	list[tuple[list[str], str]]	Keyed default table: [(key_components, default_value), ...]. key_components aligns positionally with enumeration_def_index_ids. Empty when absent.

Source code in src/cifflow/dictionary/schema.py

@dataclass
class ColumnDef:
    """
    Definition of a single column in a generated SQLite table.

    Attributes
    ----------
    name:
        SQL column name, equal to the DDLm ``_name.object_id``, lowercased.
        For synthetic columns the name is ``_cifflow_block_id``, ``_cifflow_row_id``, or
        ``_cifflow_id``.
    definition_id:
        The current canonical ``_definition.id`` for this column's DDLm item.
        Empty string for synthetic columns.
    type_contents:
        DDLm ``_type.contents`` value (e.g. ``"Text"``, ``"Integer"``,
        ``"Real"``, ``"List"``); ``None`` if absent from the dictionary or for
        synthetic columns.  Informational only -- DDL always emits ``TEXT`` for
        all value columns; ``_cifflow_row_id`` always emits ``INTEGER``.
    nullable:
        ``False`` for synthetic and primary-key columns; ``True`` for all
        other domain columns.
    is_primary_key:
        ``True`` if this column is part of the table's ``PRIMARY KEY``.
    is_synthetic:
        ``True`` for the ``_cifflow_block_id``, ``_cifflow_row_id``, and ``_cifflow_id``
        infrastructure columns, which have no corresponding DDLm item
        definition.
    linked_item_id:
        For ``SU`` items only: the ``_definition.id`` of the associated
        measurand item, lowercased.  ``None`` for all other column types.
        Does not produce a ``FOREIGN KEY`` constraint; used by the ingestion
        and output layers.
    type_container:
        DDLm ``_type.container`` value (e.g. ``"Single"``, ``"List"``,
        ``"Matrix"``); ``None`` for synthetic columns, ``"Single"`` as the
        DDLm default for domain columns when the attribute is absent.
        Non-``"Single"`` containers store JSON text in SQLite regardless of
        ``type_contents``.
    enumeration_states:
        Allowed enumeration values from ``_enumeration_set.state``.  Empty
        list when not present.
    enumeration_range:
        Value of ``_enumeration.range``.  ``None`` if absent.
    type_dimension:
        Value of ``_type.dimension``.  ``None`` if absent.
    enumeration_default:
        Scalar default value from ``_enumeration.default``.  ``None`` if absent.
    enumeration_def_index_ids:
        Ordered index tag names from ``_enumeration.def_index_ids`` for keyed
        default lookup.  Empty when absent.
    enumeration_defaults:
        Keyed default table: ``[(key_components, default_value), ...]``.
        ``key_components`` aligns positionally with ``enumeration_def_index_ids``.
        Empty when absent.
    """

    name: str
    definition_id: str
    type_contents: str | None
    nullable: bool
    is_primary_key: bool
    is_synthetic: bool
    linked_item_id: str | None
    type_container: str | None = 'Single'
    enumeration_states: list[str] = field(default_factory=list)
    enumeration_range: str | None = None
    type_dimension: str | None = None
    enumeration_default: str | None = None
    enumeration_def_index_ids: list[str] = field(default_factory=list)
    enumeration_defaults: list[tuple[list[str], str]] = field(default_factory=list)

TableDef dataclass

Definition of a single SQLite table generated from a DDLm category.

Attributes:

Name	Type	Description
name	str	SQL table name, derived from the category's _definition.id (lowercased, leading _ stripped, . replaced with _).
definition_id	str	The _definition.id of the category save frame that produced this table.
category_class	str	DDLm class of the source category: "Set" or "Loop".
columns	list[ColumnDef]	Ordered list of column definitions. Order follows the column-ordering rule: _cifflow_block_id, _cifflow_id (keyless Set only), _cifflow_row_id, primary-key domain columns, remaining domain columns alphabetically.
primary_keys	list[str]	Column names forming the PRIMARY KEY, in declaration order.
foreign_keys	list[ForeignKeyDef]	FOREIGN KEY constraints on this table; empty when none exist.

Source code in src/cifflow/dictionary/schema.py

@dataclass
class TableDef:
    """
    Definition of a single SQLite table generated from a DDLm category.

    Attributes
    ----------
    name:
        SQL table name, derived from the category's ``_definition.id``
        (lowercased, leading ``_`` stripped, ``.`` replaced with ``_``).
    definition_id:
        The ``_definition.id`` of the category save frame that produced
        this table.
    category_class:
        DDLm class of the source category: ``"Set"`` or ``"Loop"``.
    columns:
        Ordered list of column definitions.  Order follows the column-ordering
        rule: ``_cifflow_block_id``, ``_cifflow_id`` (keyless Set only),
        ``_cifflow_row_id``, primary-key domain columns, remaining domain columns
        alphabetically.
    primary_keys:
        Column names forming the ``PRIMARY KEY``, in declaration order.
    foreign_keys:
        ``FOREIGN KEY`` constraints on this table; empty when none exist.
    """

    name: str
    definition_id: str
    category_class: str
    columns: list[ColumnDef]
    primary_keys: list[str]
    foreign_keys: list[ForeignKeyDef] = field(default_factory=list)

SchemaSpec dataclass

Complete SQLite schema derived from a DdlmDictionary.

Produced by :func:generate_schema and consumed by :func:emit_create_statements and :func:~cifflow.dictionary.schema_apply.apply_schema.

Attributes:

Name	Type	Description
tables	dict[str, TableDef]	Mapping from SQL table name to its :class:TableDef.
column_to_tag	dict[tuple[str, str], str]	Reverse mapping from (table_name, column_name) to the canonical _definition.id of the corresponding DDLm item. Synthetic columns (_cifflow_block_id, _cifflow_row_id, _cifflow_id) are excluded.
alias_to_definition_id	dict[str, str]	Old tag name → canonical _definition.id. Copied from DdlmDictionary.alias_to_definition_id by generate_schema. Used by ingest() for alias resolution without retaining a dictionary reference.
deprecated_ids	set[str]	Set of _definition.id values marked as deprecated. Copied from DdlmDictionary.deprecated_ids by generate_schema. Used by ingest() to emit deprecation warnings.
warnings	list[str]	Non-fatal issues encountered during schema generation, in emission order.
bridge_columns	list[BridgeColumnDef]	Transitive bridge column definitions -- derived columns whose values are resolved through one or more FK lookup hops.
propagation_links	dict[str, list[tuple[str, str, str | None]]]	Mapping from table name to a list of (column_name, target_def_id, default) tuples for PK Link columns whose FK constraint was skipped at schema generation time.
dictionary_name	str | None	data_ block name from the source dictionary CIF (e.g. "CIF_CORE"). Copied from :attr:~cifflow.dictionary.ddlm_parser.DdlmDictionary.name.
dictionary_title	str | None	Value of _dictionary.title from the source. None if absent.
dictionary_version	str | None	Value of _dictionary.version from the source. None if absent.
dictionary_uri	str | None	Value of _dictionary.uri from the source. None if absent.
source_files	list[str]	Absolute file paths of every dictionary file loaded. Empty when no path_resolver was supplied to :class:~cifflow.dictionary.loader.DictionaryLoader.
category_parent	dict[str, str | None]	Mapping from table name to its parent table name (or None for root categories) in the DDLm category-parent hierarchy.
tag_to_category_class	dict[str, str]	Mapping from canonical _definition.id to the DDLm class ("Set" or "Loop") of the category that contains it. Covers all dictionary items, including those in categories that do not generate schema tables. Used by inspect_schema to trace Set-derived category keys transitively without requiring a live dictionary reference.
deprecated_replacements	dict[str, list[str]]	Mapping from deprecated _definition.id to the list of replacement tag names from _definition_replaced.by. An empty string in the list represents a PLACEHOLDER ("."), meaning deprecated with no named replacement. Covers both deprecated items and deprecated categories.
partial_links	list[PartialLinkDef]	DDLm Link items that could not be resolved into full FK constraints during schema generation. Recorded for use by inspection tools.

Source code in src/cifflow/dictionary/schema.py

@dataclass
class SchemaSpec:
    """
    Complete SQLite schema derived from a ``DdlmDictionary``.

    Produced by :func:`generate_schema` and consumed by
    :func:`emit_create_statements` and
    :func:`~cifflow.dictionary.schema_apply.apply_schema`.

    Attributes
    ----------
    tables:
        Mapping from SQL table name to its :class:`TableDef`.
    column_to_tag:
        Reverse mapping from ``(table_name, column_name)`` to the canonical
        ``_definition.id`` of the corresponding DDLm item.  Synthetic
        columns (``_cifflow_block_id``, ``_cifflow_row_id``, ``_cifflow_id``) are excluded.
    alias_to_definition_id:
        Old tag name → canonical ``_definition.id``.  Copied from
        ``DdlmDictionary.alias_to_definition_id`` by ``generate_schema``.
        Used by ``ingest()`` for alias resolution without retaining a
        dictionary reference.
    deprecated_ids:
        Set of ``_definition.id`` values marked as deprecated.  Copied from
        ``DdlmDictionary.deprecated_ids`` by ``generate_schema``.  Used by
        ``ingest()`` to emit deprecation warnings.
    warnings:
        Non-fatal issues encountered during schema generation, in emission
        order.
    bridge_columns:
        Transitive bridge column definitions -- derived columns whose values
        are resolved through one or more FK lookup hops.
    propagation_links:
        Mapping from table name to a list of
        ``(column_name, target_def_id, default)`` tuples for PK ``Link``
        columns whose FK constraint was skipped at schema generation time.
    dictionary_name:
        ``data_`` block name from the source dictionary CIF (e.g.
        ``"CIF_CORE"``).  Copied from
        :attr:`~cifflow.dictionary.ddlm_parser.DdlmDictionary.name`.
    dictionary_title:
        Value of ``_dictionary.title`` from the source.  ``None`` if absent.
    dictionary_version:
        Value of ``_dictionary.version`` from the source.  ``None`` if absent.
    dictionary_uri:
        Value of ``_dictionary.uri`` from the source.  ``None`` if absent.
    source_files:
        Absolute file paths of every dictionary file loaded.  Empty when no
        ``path_resolver`` was supplied to
        :class:`~cifflow.dictionary.loader.DictionaryLoader`.
    category_parent:
        Mapping from table name to its parent table name (or ``None`` for
        root categories) in the DDLm category-parent hierarchy.
    tag_to_category_class:
        Mapping from canonical ``_definition.id`` to the DDLm class
        (``"Set"`` or ``"Loop"``) of the category that contains it.
        Covers *all* dictionary items, including those in categories that do
        not generate schema tables.  Used by ``inspect_schema`` to trace
        Set-derived category keys transitively without requiring a live
        dictionary reference.
    deprecated_replacements:
        Mapping from deprecated ``_definition.id`` to the list of replacement
        tag names from ``_definition_replaced.by``.  An empty string in the
        list represents a ``PLACEHOLDER`` (``"."``), meaning deprecated with
        no named replacement.  Covers both deprecated items and deprecated
        categories.
    partial_links : list[PartialLinkDef]
        DDLm ``Link`` items that could not be resolved into full FK constraints
        during schema generation.  Recorded for use by inspection tools.
    """

    tables: dict[str, TableDef]
    column_to_tag: dict[tuple[str, str], str]
    alias_to_definition_id: dict[str, str] = field(default_factory=dict)
    deprecated_ids: set[str] = field(default_factory=set)
    warnings: list[str] = field(default_factory=list)
    bridge_columns: list[BridgeColumnDef] = field(default_factory=list)
    propagation_links: dict[str, list[tuple[str, str, str | None]]] = field(default_factory=dict)
    dictionary_name: str | None = None
    dictionary_title: str | None = None
    dictionary_version: str | None = None
    dictionary_uri: str | None = None
    source_files: list[str] = field(default_factory=list)
    category_parent: dict[str, str | None] = field(default_factory=dict)
    tag_to_category_class: dict[str, str] = field(default_factory=dict)
    deprecated_replacements: dict[str, list[str]] = field(default_factory=dict)
    partial_links: list['PartialLinkDef'] = field(default_factory=list)

    def descendants(self, root: str) -> frozenset[str]:
        """Return all table names that are *root* or a descendant of *root* in the ``category_parent`` hierarchy.

        Returns ``frozenset({root})`` if *root* has no children, or
        ``frozenset()`` if *root* is not in the schema at all.
        """
        if root not in self.tables and root not in self.category_parent.values():
            return frozenset()
        result: set[str] = {root}
        for tbl in self.tables:
            p = self.category_parent.get(tbl)
            while p is not None:
                if p == root:
                    result.add(tbl)
                    break
                p = self.category_parent.get(p)
        return frozenset(result)

descendants(root)

Return all table names that are root or a descendant of root in the category_parent hierarchy.

Returns frozenset({root}) if root has no children, or frozenset() if root is not in the schema at all.

Source code in src/cifflow/dictionary/schema.py

def descendants(self, root: str) -> frozenset[str]:
    """Return all table names that are *root* or a descendant of *root* in the ``category_parent`` hierarchy.

    Returns ``frozenset({root})`` if *root* has no children, or
    ``frozenset()`` if *root* is not in the schema at all.
    """
    if root not in self.tables and root not in self.category_parent.values():
        return frozenset()
    result: set[str] = {root}
    for tbl in self.tables:
        p = self.category_parent.get(tbl)
        while p is not None:
            if p == root:
                result.add(tbl)
                break
            p = self.category_parent.get(p)
    return frozenset(result)

generate_schema(dictionary)

Derive a :class:SchemaSpec from a loaded DdlmDictionary.

Iterates over all categories in dictionary, creating one :class:TableDef for each Set or Loop category. Head and Functions categories are silently skipped (they never appear in data instance files); any other unrecognised class emits a warning and is also skipped.

Foreign-key constraints are built in a second pass over all items whose type_purpose is "Link". SU items populate :attr:ColumnDef.linked_item_id but do not produce :class:ForeignKeyDef entries.

alias_to_definition_id and deprecated_ids are copied directly from dictionary so that ingest() can perform alias resolution and deprecation checking without retaining a reference to the dictionary.

Parameters:

Name	Type	Description	Default
dictionary	DdlmDictionary	The loaded dictionary returned by :meth:~cifflow.dictionary.loader.DictionaryLoader.load.	required

Returns:

Type	Description
SchemaSpec	The complete schema specification including all tables, column definitions, primary keys, foreign keys, the reverse column_to_tag mapping, and alias/deprecation metadata.

Source code in src/cifflow/dictionary/schema.py

def generate_schema(dictionary: DdlmDictionary) -> SchemaSpec:
    """
    Derive a :class:`SchemaSpec` from a loaded ``DdlmDictionary``.

    Iterates over all categories in *dictionary*, creating one
    :class:`TableDef` for each ``Set`` or ``Loop`` category.  ``Head`` and
    ``Functions`` categories are silently skipped (they never appear in data
    instance files); any other unrecognised class emits a warning and is also
    skipped.

    Foreign-key constraints are built in a second pass over all items whose
    ``type_purpose`` is ``"Link"``.  ``SU`` items populate
    :attr:`ColumnDef.linked_item_id` but do not produce
    :class:`ForeignKeyDef` entries.

    ``alias_to_definition_id`` and ``deprecated_ids`` are copied directly from
    *dictionary* so that ``ingest()`` can perform alias resolution and
    deprecation checking without retaining a reference to the dictionary.

    Parameters
    ----------
    dictionary:
        The loaded dictionary returned by
        :meth:`~cifflow.dictionary.loader.DictionaryLoader.load`.

    Returns
    -------
    SchemaSpec
        The complete schema specification including all tables, column
        definitions, primary keys, foreign keys, the reverse
        ``column_to_tag`` mapping, and alias/deprecation metadata.
    """
    warnings: list[str] = []
    partial_links: list[PartialLinkDef] = []

    tables, column_to_tag = _build_tables(dictionary, warnings)
    bridge_columns = _build_foreign_keys(dictionary, tables, warnings, partial_links)
    propagation_links = _build_propagation_links(dictionary, tables)
    category_parent = _build_category_parent(dictionary, tables)
    tag_to_category_class, deprecated_replacements = _build_tag_metadata(dictionary)

    return SchemaSpec(
        tables=tables,
        column_to_tag=column_to_tag,
        alias_to_definition_id=dict(dictionary.alias_to_definition_id),
        deprecated_ids=set(dictionary.deprecated_ids),
        warnings=warnings,
        bridge_columns=bridge_columns,
        propagation_links=propagation_links,
        dictionary_name=dictionary.name or None,
        dictionary_title=dictionary.title or None,
        dictionary_version=dictionary.version or None,
        dictionary_uri=dictionary.uri or None,
        source_files=list(dictionary.source_files),
        category_parent=category_parent,
        tag_to_category_class=tag_to_category_class,
        deprecated_replacements=deprecated_replacements,
        partial_links=partial_links,
    )

emit_create_statements(schema)

Render each :class:TableDef in schema as a CREATE TABLE statement.

Returns one SQL string per table in topological order (FK parents before children). The statements use CREATE TABLE IF NOT EXISTS and include inline PRIMARY KEY and FOREIGN KEY clauses. All FK constraints carry DEFERRABLE INITIALLY DEFERRED.

All value columns are declared TEXT regardless of ColumnDef.type_contents; _cifflow_row_id is always INTEGER.

Parameters:

Name	Type	Description	Default
schema	SchemaSpec	The schema specification produced by :func:generate_schema.	required

Returns:

Type	Description
list[str]	One CREATE TABLE IF NOT EXISTS ... statement per table.

Source code in src/cifflow/dictionary/schema.py

def emit_create_statements(schema: SchemaSpec) -> list[str]:
    """
    Render each :class:`TableDef` in *schema* as a ``CREATE TABLE`` statement.

    Returns one SQL string per table in topological order (FK parents before
    children).  The statements use ``CREATE TABLE IF NOT EXISTS`` and include
    inline ``PRIMARY KEY`` and ``FOREIGN KEY`` clauses.  All FK constraints
    carry ``DEFERRABLE INITIALLY DEFERRED``.

    All value columns are declared ``TEXT`` regardless of
    ``ColumnDef.type_contents``; ``_cifflow_row_id`` is always ``INTEGER``.

    Parameters
    ----------
    schema:
        The schema specification produced by :func:`generate_schema`.

    Returns
    -------
    list[str]
        One ``CREATE TABLE IF NOT EXISTS ...`` statement per table.
    """
    stmts: list[str] = []

    for table in _topo_sort_tables(schema.tables):
        parts: list[str] = []

        row_id_col = next((c for c in table.columns if c.name == '_cifflow_row_id'), None)
        for col in table.columns:
            line = f"    {_qi(col.name)}  {_ddl_type(col)}"
            if not col.nullable:
                line += "  NOT NULL"
            parts.append(line)

        pk_clause = ', '.join(_qi(k) for k in table.primary_keys)
        parts.append(f"    PRIMARY KEY ({pk_clause})")

        # Composite UNIQUE on (_cifflow_block_id, _cifflow_row_id) when _cifflow_row_id is not already
        # part of the PRIMARY KEY.
        if row_id_col is not None and not row_id_col.is_primary_key:
            parts.append(
                f"    UNIQUE ({_qi('_cifflow_block_id')}, {_qi('_cifflow_row_id')})"
            )

        for fk in table.foreign_keys:
            src_cols = ', '.join(_qi(c) for c in fk.source_columns)
            tgt_cols = ', '.join(_qi(c) for c in fk.target_columns)
            parts.append(
                f"    FOREIGN KEY ({src_cols})\n"
                f"        REFERENCES {_qi(fk.target_table)}({tgt_cols})\n"
                f"        DEFERRABLE INITIALLY DEFERRED"
            )

        body = ',\n'.join(parts)
        stmts.append(
            f"CREATE TABLE IF NOT EXISTS {_qi(table.name)} (\n{body}\n)"
        )

    return stmts

emit_fallback_create_statements()

Return the fixed DDL statements for the schema-less fallback tier.

Returns four SQL strings: CREATE TABLE IF NOT EXISTS for _cif_fallback, its lookup index, CREATE TABLE IF NOT EXISTS for _block_dataset_membership, and CREATE TABLE IF NOT EXISTS for _validation_result.

Source code in src/cifflow/dictionary/schema.py

def emit_fallback_create_statements() -> list[str]:
    """
    Return the fixed DDL statements for the schema-less fallback tier.

    Returns four SQL strings: ``CREATE TABLE IF NOT EXISTS`` for
    ``_cif_fallback``, its lookup index, ``CREATE TABLE IF NOT EXISTS`` for
    ``_block_dataset_membership``, and ``CREATE TABLE IF NOT EXISTS`` for
    ``_validation_result``.
    """
    fallback = (
        f"CREATE TABLE IF NOT EXISTS {_qi('_cif_fallback')} (\n"
        f"    {_qi('_cifflow_block_id')}   TEXT     NOT NULL,\n"
        f"    {_qi('_cifflow_row_id')}     INTEGER  NOT NULL,\n"
        f"    {_qi('tag')}         TEXT     NOT NULL,\n"
        f"    {_qi('value')}       TEXT,\n"
        f"    {_qi('value_type')}  TEXT     NOT NULL,\n"
        f"    {_qi('loop_id')}     INTEGER,\n"
        f"    {_qi('col_index')}   INTEGER,\n"
        f"    {_qi('ref_table')}   TEXT,\n"
        f"    PRIMARY KEY ({_qi('_cifflow_block_id')}, {_qi('_cifflow_row_id')}, {_qi('tag')})\n"
        f")"
    )
    index = (
        f"CREATE INDEX IF NOT EXISTS {_qi('idx_cif_fallback_tag_block')} "
        f"ON {_qi('_cif_fallback')} ({_qi('tag')}, {_qi('_cifflow_block_id')})"
    )
    membership = (
        f"CREATE TABLE IF NOT EXISTS {_qi('_block_dataset_membership')} (\n"
        f"    {_qi('_cifflow_block_id')}            TEXT  NOT NULL,\n"
        f"    {_qi('_audit_dataset_id')}    TEXT  NOT NULL,\n"
        f"    {_qi('id_regime')}            TEXT  NOT NULL,\n"
        f"    PRIMARY KEY ({_qi('_cifflow_block_id')}, {_qi('_audit_dataset_id')})\n"
        f")"
    )
    validation = (
        f"CREATE TABLE IF NOT EXISTS {_qi('_validation_result')} (\n"
        f"    {_qi('check_name')}  TEXT  NOT NULL,\n"
        f"    {_qi('severity')}    TEXT  NOT NULL,\n"
        f"    {_qi('block_id')}    TEXT,\n"
        f"    {_qi('detail')}      TEXT,\n"
        f"    {_qi('id_regime')}   TEXT\n"
        f")"
    )
    block_order = (
        f"CREATE TABLE IF NOT EXISTS {_qi('_block_order')} (\n"
        f"    {_qi('_cifflow_block_id')}   TEXT     PRIMARY KEY,\n"
        f"    {_qi('position')}    INTEGER  NOT NULL\n"
        f")"
    )
    tag_presence = (
        f"CREATE TABLE IF NOT EXISTS {_qi('_tag_presence')} (\n"
        f"    {_qi('_cifflow_block_id')}     TEXT  NOT NULL,\n"
        f"    {_qi('table_name')}    TEXT  NOT NULL,\n"
        f"    {_qi('column_name')}   TEXT  NOT NULL,\n"
        f"    {_qi('pk_json')}       TEXT  NOT NULL,\n"
        f"    PRIMARY KEY ({_qi('_cifflow_block_id')}, {_qi('table_name')}, {_qi('column_name')}, {_qi('pk_json')})\n"
        f")"
    )
    return [fallback, index, membership, validation, block_order, tag_presence]

cifflow.dictionary.resolver

Tag resolution — maps a CIF tag name to its current DDLm definition.

ResolvedTag dataclass

Result of resolving a tag name against a loaded DDLm dictionary.

Produced by :func:resolve_tag.

Attributes:

Name	Type	Description
definition_id	str	The current canonical tag name (_definition.id), lowercased.
category_id	str	The SQL table name for this definition (_name.category_id), lowercased.
object_id	str	The SQL column name for this definition (_name.object_id), lowercased.
was_alias	bool	True if the input tag was an old alias that maps to definition_id; False if it matched the canonical name directly.
is_deprecated	bool	True if this definition has been superseded by one or more replacements (_definition_replaced records exist).

Source code in src/cifflow/dictionary/resolver.py

@dataclass
class ResolvedTag:
    """
    Result of resolving a tag name against a loaded DDLm dictionary.

    Produced by :func:`resolve_tag`.

    Attributes
    ----------
    definition_id:
        The current canonical tag name (``_definition.id``), lowercased.
    category_id:
        The SQL table name for this definition (``_name.category_id``),
        lowercased.
    object_id:
        The SQL column name for this definition (``_name.object_id``),
        lowercased.
    was_alias:
        ``True`` if the input tag was an old alias that maps to
        *definition_id*; ``False`` if it matched the canonical name directly.
    is_deprecated:
        ``True`` if this definition has been superseded by one or more
        replacements (``_definition_replaced`` records exist).
    """

    definition_id: str
    category_id: str
    object_id: str
    was_alias: bool
    is_deprecated: bool

resolve_tag(tag, dictionary)

Resolve a tag name from a CIF data file to its current definition.

Looks up tag (case-insensitive) in dictionary.tag_to_item, following alias chains transparently. Returns None if the tag is not known to this dictionary; this is the signal that the tag belongs to the fallback tier, not an error condition.

Does not emit warnings. The caller is responsible for acting on the was_alias and is_deprecated flags of the returned value.

Parameters:

Name	Type	Description	Default
tag	str	The tag name to resolve, as it appears in a CIF data file. Lookup is case-insensitive.	required
dictionary	DdlmDictionary	The loaded DdlmDictionary to resolve against.	required

Returns:

Type	Description
ResolvedTag | None	Resolution result including canonical name, table, column, and alias/deprecation flags; None if the tag is not known to this dictionary.

Source code in src/cifflow/dictionary/resolver.py

def resolve_tag(tag: str, dictionary: DdlmDictionary) -> ResolvedTag | None:
    """
    Resolve a tag name from a CIF data file to its current definition.

    Looks up *tag* (case-insensitive) in ``dictionary.tag_to_item``,
    following alias chains transparently.  Returns ``None`` if the tag is
    not known to this dictionary; this is the signal that the tag belongs
    to the fallback tier, not an error condition.

    Does not emit warnings.  The caller is responsible for acting on the
    ``was_alias`` and ``is_deprecated`` flags of the returned value.

    Parameters
    ----------
    tag:
        The tag name to resolve, as it appears in a CIF data file.
        Lookup is case-insensitive.
    dictionary:
        The loaded ``DdlmDictionary`` to resolve against.

    Returns
    -------
    ResolvedTag | None
        Resolution result including canonical name, table, column, and
        alias/deprecation flags; ``None`` if the tag is not known to this
        dictionary.
    """
    item = dictionary.tag_to_item.get(tag.lower())
    if item is None:
        return None

    was_alias = tag.lower() in dictionary.alias_to_definition_id

    if item.category_id is None or item.object_id is None:
        return None

    return ResolvedTag(
        definition_id=item.definition_id,
        category_id=item.category_id,
        object_id=item.object_id,
        was_alias=was_alias,
        is_deprecated=item.is_deprecated,
    )

cifflow.dictionary.cache

JSON serialisation and deserialisation of DdlmDictionary.

Allows a fully resolved dictionary (including metadictionary imports) to be saved to disk and reloaded without re-parsing constituent CIF files.

Cache invalidation is the caller's responsibility. These functions make no attempt to detect whether the source dictionary files have changed.

save_dictionary(dictionary, path)

Serialise dictionary to a JSON file at path.

The file is written atomically from the perspective of a single process (standard open + json.dump). Existing files are overwritten.

tag_to_item is stored as a dict[str, str] mapping (tag name → definition_id) to avoid duplicating DdlmItem objects for every alias. It is reconstructed on load.

Parameters:

Name	Type	Description	Default
dictionary	DdlmDictionary	The DdlmDictionary to serialise.	required
path	str | Path	Destination file path. Parent directories must already exist.	required

Source code in src/cifflow/dictionary/cache.py

def save_dictionary(
    dictionary: DdlmDictionary,
    path: str | pathlib.Path,
) -> None:
    """
    Serialise *dictionary* to a JSON file at *path*.

    The file is written atomically from the perspective of a single process
    (standard ``open`` + ``json.dump``).  Existing files are overwritten.

    ``tag_to_item`` is stored as a ``dict[str, str]`` mapping (tag name →
    ``definition_id``) to avoid duplicating ``DdlmItem`` objects for every
    alias.  It is reconstructed on load.

    Parameters
    ----------
    dictionary:
        The ``DdlmDictionary`` to serialise.
    path:
        Destination file path.  Parent directories must already exist.
    """
    data = {
        'name': dictionary.name,
        'title': dictionary.title,
        'version': dictionary.version,
        'uri': dictionary.uri,
        'categories': {
            k: dataclasses.asdict(v)
            for k, v in dictionary.categories.items()
        },
        'items': {
            k: dataclasses.asdict(v)
            for k, v in dictionary.items.items()
        },
        # Store as tag → definition_id to avoid duplicating DdlmItem objects.
        'tag_to_item': {
            tag: item.definition_id
            for tag, item in dictionary.tag_to_item.items()
        },
        'alias_to_definition_id': dictionary.alias_to_definition_id,
        'deprecated_ids': sorted(dictionary.deprecated_ids),
        'warnings': dictionary.warnings,
        'source_files': dictionary.source_files,
    }
    with open(path, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

load_dictionary(path)

Deserialise a DdlmDictionary from a JSON file at path.

Raises ValueError if the file does not exist, contains malformed JSON, is missing required keys, or references an unknown definition_id in tag_to_item. The caller should respond by falling back to DictionaryLoader.load().

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a JSON file previously written by :func:save_dictionary.	required

Returns:

Type	Description
DdlmDictionary	The deserialised dictionary.

Raises:

Type	Description
ValueError	If the file cannot be read or the contents are invalid.

Source code in src/cifflow/dictionary/cache.py

def load_dictionary(path: str | pathlib.Path) -> DdlmDictionary:
    """
    Deserialise a ``DdlmDictionary`` from a JSON file at *path*.

    Raises ``ValueError`` if the file does not exist, contains malformed JSON,
    is missing required keys, or references an unknown ``definition_id`` in
    ``tag_to_item``.  The caller should respond by falling back to
    ``DictionaryLoader.load()``.

    Parameters
    ----------
    path:
        Path to a JSON file previously written by :func:`save_dictionary`.

    Returns
    -------
    DdlmDictionary
        The deserialised dictionary.

    Raises
    ------
    ValueError
        If the file cannot be read or the contents are invalid.
    """
    try:
        with open(path, 'r', encoding='utf-8') as f:
            data = json.load(f)
    except FileNotFoundError:
        raise ValueError(f'dictionary cache file not found: {path}')
    except json.JSONDecodeError as e:
        raise ValueError(f'malformed JSON in dictionary cache {path}: {e}')

    try:
        categories = {
            k: DdlmItem(**v) for k, v in data['categories'].items()
        }
        items = {
            k: DdlmItem(**v) for k, v in data['items'].items()
        }
    except (KeyError, TypeError) as e:
        raise ValueError(f'invalid dictionary cache structure in {path}: {e}')

    # Reconstruct tag_to_item from the stored tag → definition_id mapping.
    all_by_id: dict[str, DdlmItem] = {**categories, **items}
    tag_to_item: dict[str, DdlmItem] = {}
    try:
        for tag, def_id in data['tag_to_item'].items():
            if def_id not in all_by_id:
                raise ValueError(
                    f'tag_to_item entry {tag!r} references unknown '
                    f'definition_id {def_id!r} in {path}'
                )
            tag_to_item[tag] = all_by_id[def_id]
    except (KeyError, TypeError) as e:
        raise ValueError(f'invalid tag_to_item in dictionary cache {path}: {e}')

    try:
        return DdlmDictionary(
            name=data['name'],
            title=data['title'],
            version=data['version'],
            categories=categories,
            items=items,
            tag_to_item=tag_to_item,
            alias_to_definition_id=data['alias_to_definition_id'],
            deprecated_ids=set(data['deprecated_ids']),
            warnings=data['warnings'],
            source_files=data.get('source_files', []),
            uri=data.get('uri'),
        )
    except (KeyError, TypeError) as e:
        raise ValueError(f'invalid dictionary cache structure in {path}: {e}')

cifflow.dictionary.visualise

Schema visualisation: DOT and HTML output for SchemaSpec.

Public functions

visualise_schema(schema, ...) -> Graphviz DOT string visualise_schema_html(schema, ...) -> self-contained HTML string

visualise_schema(schema, *, show_columns='sparse', show_bridge=True, show_parent_edges=True, highlight_orphans=True, highlight_components=False, show_orphans=True, show_legend=True, concentrate=False, hide_deprecated=False, layout='dot', splines='curved', ranksep=1.0, nodesep=0.4)

Return a Graphviz DOT string visualising schema.

Parameters:

Name	Type	Description	Default
schema	SchemaSpec	The :class:~cifflow.dictionary.schema.SchemaSpec to visualise.	required
show_columns	Literal['all', 'sparse', 'none']	'all' — every column; 'sparse' — only PK and key/bridge columns; 'none' — header only.	'sparse'
show_bridge	bool	Include bridge column edges. Always True for [BRIDGE ONLY] nodes.	True
show_parent_edges	bool	Include category-parent hierarchy edges. Always True when the target is a ghost node.	True
highlight_orphans	bool	Apply [ORPHAN] / [BRIDGE ONLY] badges and border styles.	True
highlight_components	bool	Wrap each connected component in a subgraph cluster_ box.	False
show_orphans	bool	When False, [ORPHAN] and [BRIDGE ONLY] nodes (and their edges) are omitted entirely.	True
show_legend	bool	When True (default), emit a __legend__ node summarising node colours, connectivity badges, edge styles, and column badges. The content of the legend adapts to the active flags.	True
concentrate	bool	When True, set concentrate=true in the graph attributes. Graphviz merges parallel edges that share a common endpoint into a shared spine, reducing visual clutter in dense schemas.	False
hide_deprecated	bool	When True, deprecated columns (those whose definition_id appears in schema.deprecated_ids) are omitted from column rows. Any table where every non-synthetic column is deprecated is removed from the graph entirely — no node, no ghost, no edges.	False
layout	str	Graphviz layout engine written into graph [layout=...]. viz.js reads this attribute automatically.	'dot'
splines	str	Graphviz splines attribute controlling edge routing. 'curved' (default) draws smooth distinct arcs and handles edge labels correctly, including edges that run backwards in the layout. 'ortho' routes edges as right-angle lines but has known issues with label placement and backwards edges. Other values: 'polyline', 'spline', 'none'.	'curved'
ranksep	float	Minimum separation in inches between ranks (layout rows/columns). Larger values spread the graph out vertically (or horizontally with rankdir=LR) and give edge routing more room. Default 1.0.	1.0
nodesep	float	Minimum separation in inches between adjacent nodes in the same rank. Default 0.4.	0.4

Returns:

Type	Description
str	A Graphviz DOT source string ready to pass to a Graphviz renderer or embed in an HTML page via viz.js.

Source code in src/cifflow/dictionary/visualise.py

def visualise_schema(
    schema: SchemaSpec,
    *,
    show_columns: Literal['all', 'sparse', 'none'] = 'sparse',
    show_bridge: bool = True,
    show_parent_edges: bool = True,
    highlight_orphans: bool = True,
    highlight_components: bool = False,
    show_orphans: bool = True,
    show_legend: bool = True,
    concentrate: bool = False,
    hide_deprecated: bool = False,
    layout: str = 'dot',
    splines: str = 'curved',
    ranksep: float = 1.0,
    nodesep: float = 0.4,
) -> str:
    """
    Return a Graphviz DOT string visualising *schema*.

    Parameters
    ----------
    schema:
        The :class:`~cifflow.dictionary.schema.SchemaSpec` to visualise.
    show_columns:
        ``'all'`` — every column; ``'sparse'`` — only PK and key/bridge columns;
        ``'none'`` — header only.
    show_bridge:
        Include bridge column edges.  Always ``True`` for ``[BRIDGE ONLY]`` nodes.
    show_parent_edges:
        Include category-parent hierarchy edges.  Always ``True`` when the target
        is a ghost node.
    highlight_orphans:
        Apply ``[ORPHAN]`` / ``[BRIDGE ONLY]`` badges and border styles.
    highlight_components:
        Wrap each connected component in a ``subgraph cluster_`` box.
    show_orphans:
        When ``False``, ``[ORPHAN]`` and ``[BRIDGE ONLY]`` nodes (and their edges)
        are omitted entirely.
    show_legend:
        When ``True`` (default), emit a ``__legend__`` node summarising node
        colours, connectivity badges, edge styles, and column badges.  The
        content of the legend adapts to the active flags.
    concentrate:
        When ``True``, set ``concentrate=true`` in the graph attributes.
        Graphviz merges parallel edges that share a common endpoint into a
        shared spine, reducing visual clutter in dense schemas.
    hide_deprecated:
        When ``True``, deprecated columns (those whose ``definition_id``
        appears in ``schema.deprecated_ids``) are omitted from column rows.
        Any table where every non-synthetic column is deprecated is removed
        from the graph entirely — no node, no ghost, no edges.
    layout:
        Graphviz layout engine written into ``graph [layout=...]``.  viz.js
        reads this attribute automatically.
    splines:
        Graphviz ``splines`` attribute controlling edge routing.  ``'curved'``
        (default) draws smooth distinct arcs and handles edge labels correctly,
        including edges that run backwards in the layout.  ``'ortho'`` routes
        edges as right-angle lines but has known issues with label placement
        and backwards edges.  Other values: ``'polyline'``, ``'spline'``,
        ``'none'``.
    ranksep:
        Minimum separation in inches between ranks (layout rows/columns).
        Larger values spread the graph out vertically (or horizontally with
        ``rankdir=LR``) and give edge routing more room.  Default ``1.0``.
    nodesep:
        Minimum separation in inches between adjacent nodes in the same rank.
        Default ``0.4``.

    Returns
    -------
    str
        A Graphviz DOT source string ready to pass to a Graphviz renderer or
        embed in an HTML page via viz.js.
    """
    ghost_tables = _collect_ghost_tables(schema)
    bridge_only, orphans, pass1_components = _classify_tables(schema)
    deprecated_ids, real_tables, ghost_tables = _build_vis_context(
        schema, hide_deprecated, show_orphans, bridge_only, orphans, ghost_tables
    )

    concentrate_attr = ' concentrate=true' if concentrate else ''
    lines: list[str] = [
        'digraph schema {',
        f'    graph [rankdir=LR layout="{_escape(layout)}" splines="{_escape(splines)}"'
        f' ranksep={ranksep} nodesep={nodesep}'
        f' fontname="Helvetica" fontsize=11{concentrate_attr}]',
        '    node  [fontname="Helvetica" fontsize=10]',
        '    edge  [fontname="Helvetica" fontsize=9 decorate=true]',
        '',
    ]

    # --- Connectivity lookup (used by flat path) ---
    def _connectivity(name: str) -> str:
        if name in bridge_only:
            return 'bridge_only'
        if name in orphans:
            return 'orphan'
        return 'connected'

    # --- Ghost nodes ---
    if ghost_tables:
        for ghost in sorted(ghost_tables):
            lines += _ghost_node_dot(ghost)
        lines.append('')

    # --- Real table nodes (possibly clustered) ---
    if highlight_components:
        _emit_clustered_nodes(
            lines, schema, real_tables, ghost_tables, bridge_only, orphans,
            pass1_components, highlight_orphans, show_columns, deprecated_ids,
        )
    else:
        for tbl_name in sorted(real_tables):
            tbl = schema.tables[tbl_name]
            lines += _table_node_dot(tbl, _connectivity(tbl_name), highlight_orphans, show_columns, schema, deprecated_ids)
            lines.append('')

    # --- Legend node ---
    if show_legend:
        lines += _legend_dot(highlight_orphans, show_bridge, show_parent_edges, show_columns)
        lines.append('')

    # --- Edges ---
    lines.append('')
    _emit_fk_edges(lines, schema, real_tables, ghost_tables, show_columns, deprecated_ids)
    _emit_bridge_edges(lines, schema, real_tables, ghost_tables, bridge_only, show_bridge)
    _emit_parent_edges(lines, schema, real_tables, ghost_tables, show_parent_edges)

    lines.append('}')
    return '\n'.join(lines)

visualise_schema_html(schema, *, title=None, show_columns='sparse', show_bridge=True, show_parent_edges=True, highlight_orphans=True, highlight_components=False, show_orphans=True, show_legend=True, concentrate=False, hide_deprecated=False, layout='dot', splines='curved', ranksep=1.0, nodesep=0.4)

Return a self-contained HTML string that renders schema interactively.

All keyword arguments except title are forwarded to :func:visualise_schema. The returned HTML embeds viz.js and svg-pan-zoom as inline <script> blocks — no network access is required.

Parameters:

Name	Type	Description	Default
schema	SchemaSpec	The :class:~cifflow.dictionary.schema.SchemaSpec to render.	required
title	str | None	<title> element text. Defaults to schema.dictionary_name or 'Schema' when not given.	None
show_columns	Literal['all', 'sparse', 'none']	'all', 'sparse', or 'none'. Forwarded to :func:visualise_schema.	'sparse'
show_bridge	bool	Forwarded to :func:visualise_schema.	True
show_parent_edges	bool	Forwarded to :func:visualise_schema.	True
highlight_orphans	bool	Forwarded to :func:visualise_schema.	True
highlight_components	bool	Forwarded to :func:visualise_schema.	False
show_orphans	bool	Forwarded to :func:visualise_schema.	True
show_legend	bool	Forwarded to :func:visualise_schema.	True
concentrate	bool	Forwarded to :func:visualise_schema.	False
hide_deprecated	bool	Forwarded to :func:visualise_schema.	False
layout	str	Forwarded to :func:visualise_schema.	'dot'
splines	str	Forwarded to :func:visualise_schema.	'curved'
ranksep	float	Forwarded to :func:visualise_schema.	1.0
nodesep	float	Forwarded to :func:visualise_schema.	0.4

Returns:

Type	Description
str	A self-contained HTML document with the schema rendered as an interactive SVG via viz.js and svg-pan-zoom.

Source code in src/cifflow/dictionary/visualise.py

def visualise_schema_html(
    schema: SchemaSpec,
    *,
    title: str | None = None,
    show_columns: Literal['all', 'sparse', 'none'] = 'sparse',
    show_bridge: bool = True,
    show_parent_edges: bool = True,
    highlight_orphans: bool = True,
    highlight_components: bool = False,
    show_orphans: bool = True,
    show_legend: bool = True,
    concentrate: bool = False,
    hide_deprecated: bool = False,
    layout: str = 'dot',
    splines: str = 'curved',
    ranksep: float = 1.0,
    nodesep: float = 0.4,
) -> str:
    """
    Return a self-contained HTML string that renders *schema* interactively.

    All keyword arguments except *title* are forwarded to
    :func:`visualise_schema`.  The returned HTML embeds viz.js and svg-pan-zoom
    as inline ``<script>`` blocks — no network access is required.

    Parameters
    ----------
    schema
        The :class:`~cifflow.dictionary.schema.SchemaSpec` to render.
    title
        ``<title>`` element text.  Defaults to ``schema.dictionary_name``
        or ``'Schema'`` when not given.
    show_columns
        ``'all'``, ``'sparse'``, or ``'none'``.  Forwarded to :func:`visualise_schema`.
    show_bridge
        Forwarded to :func:`visualise_schema`.
    show_parent_edges
        Forwarded to :func:`visualise_schema`.
    highlight_orphans
        Forwarded to :func:`visualise_schema`.
    highlight_components
        Forwarded to :func:`visualise_schema`.
    show_orphans
        Forwarded to :func:`visualise_schema`.
    show_legend
        Forwarded to :func:`visualise_schema`.
    concentrate
        Forwarded to :func:`visualise_schema`.
    hide_deprecated
        Forwarded to :func:`visualise_schema`.
    layout
        Forwarded to :func:`visualise_schema`.
    splines
        Forwarded to :func:`visualise_schema`.
    ranksep
        Forwarded to :func:`visualise_schema`.
    nodesep
        Forwarded to :func:`visualise_schema`.

    Returns
    -------
    str
        A self-contained HTML document with the schema rendered as an
        interactive SVG via viz.js and svg-pan-zoom.
    """
    dot_string = visualise_schema(
        schema,
        show_columns=show_columns,
        show_bridge=show_bridge,
        show_parent_edges=show_parent_edges,
        highlight_orphans=highlight_orphans,
        highlight_components=highlight_components,
        show_orphans=show_orphans,
        show_legend=show_legend,
        concentrate=concentrate,
        hide_deprecated=hide_deprecated,
        layout=layout,
        splines=splines,
        ranksep=ranksep,
        nodesep=nodesep,
    )

    page_title = title or schema.dictionary_name or 'Schema'
    page_title_escaped = html.escape(page_title)

    viz_js = _read_js('viz.js')
    full_render_js = _read_js('full.render.js')
    svg_pan_zoom_js = _read_js('svg-pan-zoom.min.js')

    # Escape DOT string for embedding in a JS template literal
    dot_escaped = dot_string.replace('\\', '\\\\').replace('`', '\\`').replace('${', '\\${')

    return f"""<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  <title>{page_title_escaped}</title>
  <script>{viz_js}</script>
  <script>{full_render_js}</script>
  <script>{svg_pan_zoom_js}</script>
  <style>
    body {{ margin: 0; background: #fafafa; }}
    #graph {{ width: 100vw; height: 100vh; overflow: hidden; }}
    #graph svg {{ width: 100%; height: 100%; }}
  </style>
</head>
<body>
  <div id="graph"></div>
  <script>
    const dot = `{dot_escaped}`;
    new Viz().renderSVGElement(dot).then(svg => {{
      document.getElementById('graph').appendChild(svg);
      svgPanZoom(svg, {{ zoomEnabled: true, controlIconsEnabled: true, fit: true, center: true }});
    }}).catch(err => {{
      document.getElementById('graph').textContent = 'Render error: ' + err;
    }});
  </script>
</body>
</html>"""