Inspect

cifflow.inspect._lexer

inspect_lexer — pretty-print the lexer token stream for a CIF source.

inspect_lexer(source, *, version=None, file=sys.stdout, use_colour=True)

Print the full token stream for source to file.

Parameters:

Name	Type	Description	Default
source	_Source	CIF source: a raw string, a pathlib.Path, or an open text file object.	required
version	Optional[CifVersion]	If None (default), auto-detected from the magic line.	None
file	TextIO	Output stream (default sys.stdout).	stdout
use_colour	bool	If False, suppress all ANSI colour codes regardless of terminal type. Default True.	True

Source code in src/cifflow/inspect/_lexer.py

def inspect_lexer(
    source: _Source,
    *,
    version: Optional[CifVersion] = None,
    file: TextIO = sys.stdout,
    use_colour: bool = True,
) -> None:
    """Print the full token stream for *source* to *file*.

    Parameters
    ----------
    source:
        CIF source: a raw string, a ``pathlib.Path``, or an open text file object.
    version:
        If None (default), auto-detected from the magic line.
    file:
        Output stream (default ``sys.stdout``).
    use_colour:
        If False, suppress all ANSI colour codes regardless of terminal type.
        Default True.
    """
    from cifflow import cifflow_core
    from cifflow.parser.version import detect_version

    source = resolve_source(source)

    mode = None
    if version is CifVersion.CIF_1_1:
        mode = 'cif1'
    elif version is CifVersion.CIF_2_0:
        mode = 'cif2'
    else:
        # Run Python detect_version solely to surface version errors in output.
        _ver, _rem, _off, v_errors = detect_version(source)
        if v_errors:
            for ve in v_errors:
                print(
                    c(f'[VERSION ERROR] line {ve.line}: {ve.message}', RED, BOLD,
                      file=file, use_colour=use_colour),
                    file=file,
                )

    tokens, detected_version = cifflow_core.lex_cif(source, mode)

    ver_label = detected_version.value
    print(
        c(f'-- token stream  (CIF {ver_label}) --', BOLD, DIM,
          file=file, use_colour=use_colour),
        file=file,
    )
    print(
        c(
            f"{'line':>5} {'col':>4}  {'token_type':<10}  {'value_type':<22}  value",
            DIM, file=file, use_colour=use_colour,
        ),
        file=file,
    )
    print(c('-' * 72, DIM, file=file, use_colour=use_colour), file=file)

    for tok in tokens:
        vtype = tok['value_type'].value if tok['value_type'] is not None else ''
        raw   = repr(tok['value'])
        if len(raw) > 50:
            raw = raw[:47] + '…' + raw[-1]

        line_part  = c(f'{tok["line"]:>5} {tok["column"]:>4}', DIM,
                       file=file, use_colour=use_colour)
        type_part  = c(f'{tok["token_type"].value:<10}', CYAN,
                       file=file, use_colour=use_colour)
        vtype_part = c(f'{vtype:<22}', BLUE, file=file, use_colour=use_colour)
        val_part   = c(raw, GREEN if tok['token_type'].value == 'value' else YELLOW,
                       file=file, use_colour=use_colour)

        print(f'  {line_part}  {type_part}  {vtype_part}  {val_part}', file=file)

        for err in tok['errors']:
            print(
                c(
                    f'         ^ LEX ERROR  col {err["column"]}: {err["message"]}',
                    RED, file=file, use_colour=use_colour,
                ),
                file=file,
            )

    print(file=file)

cifflow.inspect._parser

inspect_parse + ParseHandler — pretty-print parser events for a CIF source.

ParseHandler

A CifParserEvents implementation that prints every event and error.

Pass an optional inner handler to forward all events after printing.

Parameters:

Name	Type	Description	Default
inner	Optional[CifParserEvents]	Optional downstream handler. All events are forwarded to it after being printed.	None
file	TextIO	Output stream (default sys.stdout).	stdout
show_values	bool	If False, add_value calls are printed as a short summary rather than one line each. Useful for large loop tables. Default True.	True
use_colour	bool	If False, suppress all ANSI colour codes regardless of terminal type. Default True.	True

Source code in src/cifflow/inspect/_parser.py

class ParseHandler:
    """A ``CifParserEvents`` implementation that prints every event and error.

    Pass an optional *inner* handler to forward all events after printing.

    Parameters
    ----------
    inner:
        Optional downstream handler.  All events are forwarded to it after
        being printed.
    file:
        Output stream (default ``sys.stdout``).
    show_values:
        If False, ``add_value`` calls are printed as a short summary rather
        than one line each.  Useful for large loop tables.  Default True.
    use_colour:
        If False, suppress all ANSI colour codes regardless of terminal type.
        Default True.
    """

    def __init__(
        self,
        inner: Optional[CifParserEvents] = None,
        *,
        file: TextIO = sys.stdout,
        show_values: bool = True,
        use_colour: bool = True,
    ) -> None:
        self._inner       = inner
        self._file        = file
        self._show_values = show_values
        self._use_colour  = use_colour
        self._depth       = 0

        print(
            c('-- parser events --', BOLD, DIM, file=self._file, use_colour=use_colour),
            file=self._file,
        )

    # -- helpers ---------------------------------------------------------------

    def _indent(self) -> str:
        return '  ' * self._depth

    def _print(self, text: str, colour: str = '') -> None:
        prefix = c(self._indent(), DIM, file=self._file, use_colour=self._use_colour)
        body   = c(text, colour, file=self._file, use_colour=self._use_colour) if colour else text
        print(prefix + body, file=self._file)

    def _fwd(self, name: str, *args, **kwargs) -> None:
        if self._inner is not None:
            getattr(self._inner, name)(*args, **kwargs)

    # -- CifParserEvents -------------------------------------------------------

    def on_data_block(self, name: str) -> None:
        self._depth = 0
        self._print(f'on_data_block({name!r})', BOLD)
        self._depth = 1
        self._fwd('on_data_block', name)

    def on_save_frame_start(self, name: str) -> None:
        self._print(f'on_save_frame_start({name!r})', CYAN)
        self._depth += 1
        self._fwd('on_save_frame_start', name)

    def on_save_frame_end(self) -> None:
        self._depth = max(1, self._depth - 1)
        self._print('on_save_frame_end()', CYAN)
        self._fwd('on_save_frame_end')

    def add_tag(self, tag_name: str) -> None:
        self._print(f'add_tag({tag_name!r})', YELLOW)
        self._fwd('add_tag', tag_name)

    def add_value(self, value: str, value_type: ValueType) -> None:
        if self._show_values:
            raw = repr(value)
            if len(raw) > 60:
                raw = raw[:57] + '…' + raw[-1]
            self._print(f'add_value({raw}, {value_type.value})', GREEN)
        self._fwd('add_value', value, value_type)

    def on_list_start(self) -> None:
        self._print('on_list_start()', MAGENTA)
        self._depth += 1
        self._fwd('on_list_start')

    def on_list_end(self) -> None:
        self._depth = max(0, self._depth - 1)
        self._print('on_list_end()', MAGENTA)
        self._fwd('on_list_end')

    def on_table_start(self) -> None:
        self._print('on_table_start()', MAGENTA)
        self._depth += 1
        self._fwd('on_table_start')

    def on_table_end(self) -> None:
        self._depth = max(0, self._depth - 1)
        self._print('on_table_end()', MAGENTA)
        self._fwd('on_table_end')

    def on_table_key(self, key: str, value_type: ValueType) -> None:
        self._print(f'on_table_key({key!r}, {value_type.value})', BLUE)
        self._fwd('on_table_key', key, value_type)

    def on_loop_start(self, tags: List[str]) -> None:
        self._print(f'on_loop_start({tags!r})', CYAN)
        self._depth += 1
        self._fwd('on_loop_start', tags)

    def on_loop_end(self) -> None:
        self._depth = max(1, self._depth - 1)
        self._print('on_loop_end()', CYAN)
        self._fwd('on_loop_end')

    def on_error(self, error: ParseError) -> None:
        msg = (
            f'[{error.error_type.upper()}] '
            f'line {error.line} col {error.column}: '
            f'{error.message}'
        )
        if error.context:
            msg += f'  (context: {error.context!r})'
        if error.recovery_action:
            msg += f'  -> {error.recovery_action}'
        self._print(msg, RED)
        self._fwd('on_error', error)

inspect_parse(source, *, inner=None, file=sys.stdout, show_values=True, show_tokens=True, use_colour=True)

Run the full pipeline and print token stream then parser events.

Parameters:

Name	Type	Description	Default
source	_Source	CIF source: a raw string, a pathlib.Path, or an open text file object.	required
inner	Optional[CifParserEvents]	Optional downstream handler to receive all events.	None
file	TextIO	Output stream (default sys.stdout).	stdout
show_values	bool	Forward to ParseHandler; set False to suppress add_value lines for large files.	True
show_tokens	bool	If True (default), also print the lexer token stream before events.	True
use_colour	bool	If False, suppress all ANSI colour codes regardless of terminal type. Default True.	True

Source code in src/cifflow/inspect/_parser.py

def inspect_parse(
    source: _Source,
    *,
    inner: Optional[CifParserEvents] = None,
    file: TextIO = sys.stdout,
    show_values: bool = True,
    show_tokens: bool = True,
    use_colour: bool = True,
) -> None:
    """Run the full pipeline and print token stream then parser events.

    Parameters
    ----------
    source:
        CIF source: a raw string, a ``pathlib.Path``, or an open text file object.
    inner:
        Optional downstream handler to receive all events.
    file:
        Output stream (default ``sys.stdout``).
    show_values:
        Forward to ``ParseHandler``; set False to suppress ``add_value`` lines
        for large files.
    show_tokens:
        If True (default), also print the lexer token stream before events.
    use_colour:
        If False, suppress all ANSI colour codes regardless of terminal type.
        Default True.
    """
    source = resolve_source(source)
    if show_tokens:
        inspect_lexer(source, file=file, use_colour=use_colour)

    from cifflow import cifflow_core
    handler = ParseHandler(inner, file=file, show_values=show_values, use_colour=use_colour)
    cifflow_core.parse(source, handler)
    print(file=file)

cifflow.inspect._model

inspect_model — pretty-print a CifFile or CIF source string.

inspect_model(source, *, mode='pad', file=sys.stdout, show_values=True, show_tokens=True, use_colour=True)

Run the full pipeline through the CIF model and print a summary.

Prints (in order): token stream, parser events, CifFile summary, errors.

Parameters:

Name	Type	Description	Default
source	_Source	CIF source: a raw string, a pathlib.Path, or an open text file object.	required
mode	str	Loop row-count mismatch mode passed to CifBuilder: 'pad' (default) or 'strict'.	'pad'
file	TextIO	Output stream (default sys.stdout).	stdout
show_values	bool	Forward to ParseHandler; set False to suppress add_value lines.	True
show_tokens	bool	If True (default), also print the lexer token stream before events.	True
use_colour	bool	If False, suppress all ANSI colour codes regardless of terminal type. Default True.	True

Source code in src/cifflow/inspect/_model.py

def inspect_model(
    source: _Source,
    *,
    mode: str = 'pad',
    file: TextIO = sys.stdout,
    show_values: bool = True,
    show_tokens: bool = True,
    use_colour: bool = True,
) -> None:
    """Run the full pipeline through the CIF model and print a summary.

    Prints (in order): token stream, parser events, CifFile summary, errors.

    Parameters
    ----------
    source:
        CIF source: a raw string, a ``pathlib.Path``, or an open text file object.
    mode:
        Loop row-count mismatch mode passed to ``CifBuilder``: ``'pad'``
        (default) or ``'strict'``.
    file:
        Output stream (default ``sys.stdout``).
    show_values:
        Forward to ``ParseHandler``; set False to suppress ``add_value`` lines.
    show_tokens:
        If True (default), also print the lexer token stream before events.
    use_colour:
        If False, suppress all ANSI colour codes regardless of terminal type.
        Default True.
    """
    from cifflow.cifmodel.builder import CifBuilder

    source = resolve_source(source)

    if show_tokens:
        inspect_lexer(source, file=file, use_colour=use_colour)

    errors: list[ParseError] = []
    from cifflow import cifflow_core
    builder = CifBuilder(on_error=errors.append, mode=mode)
    handler = ParseHandler(builder, file=file, show_values=show_values, use_colour=use_colour)
    cifflow_core.parse(source, handler)
    print(file=file)

    _print_model(builder.result, file=file, use_colour=use_colour)

    if errors:
        print(c('-- errors --', BOLD, DIM, file=file, use_colour=use_colour), file=file)
        for err in errors:
            loc  = c(f'line {err.line} col {err.column}', DIM, file=file, use_colour=use_colour)
            kind = c(f'[{err.error_type.upper()}]', RED, BOLD, file=file, use_colour=use_colour)
            print(f'  {kind}  {loc}  {err.message}', file=file)
            if err.recovery_action:
                print(f'    {c("->", DIM, file=file, use_colour=use_colour)} {err.recovery_action}',
                      file=file)
        print(file=file)

cifflow.inspect._schema

inspect_schema — pretty-print a SchemaSpec derived from a DDLm dictionary.

inspect_schema(source, *, show_ddl=False, file=sys.stdout, use_colour=True)

Print a structured summary of a SchemaSpec to file.

source may be:

• A :class:~cifflow.dictionary.schema.SchemaSpec — used directly.
• A :class:~cifflow.dictionary.loader.DdlmDictionary — schema generated from it.
• A pathlib.Path to a DDLm dictionary file — loaded via :class:~cifflow.dictionary.loader.DictionaryLoader with directory_resolver(path.parent) so _import.get directives resolve from the same directory.
• A raw CIF source string — parsed with no resolver (imports that require external files are silently skipped).

Parameters:

Name	Type	Description	Default
source	Union[str, Path, SchemaSpec, DdlmDictionary]	Dictionary source, a pre-built DdlmDictionary, or a SchemaSpec.	required
show_ddl	bool	If True, append the raw CREATE TABLE DDL under each table. Default False.	False
file	TextIO	Output stream. Default sys.stdout.	stdout
use_colour	bool	If False, suppress all ANSI colour codes regardless of terminal type. Default True.	True

Source code in src/cifflow/inspect/_schema.py

def inspect_schema(
    source: 'Union[str, pathlib.Path, SchemaSpec, DdlmDictionary]',
    *,
    show_ddl: bool = False,
    file: TextIO = sys.stdout,
    use_colour: bool = True,
) -> None:
    """Print a structured summary of a ``SchemaSpec`` to *file*.

    *source* may be:

    - A :class:`~cifflow.dictionary.schema.SchemaSpec` — used directly.
    - A :class:`~cifflow.dictionary.loader.DdlmDictionary` — schema generated from it.
    - A ``pathlib.Path`` to a DDLm dictionary file — loaded via
      :class:`~cifflow.dictionary.loader.DictionaryLoader` with
      ``directory_resolver(path.parent)`` so ``_import.get`` directives
      resolve from the same directory.
    - A raw CIF source string — parsed with no resolver (imports that require
      external files are silently skipped).

    Parameters
    ----------
    source:
        Dictionary source, a pre-built ``DdlmDictionary``, or a ``SchemaSpec``.
    show_ddl:
        If ``True``, append the raw ``CREATE TABLE`` DDL under each table.
        Default ``False``.
    file:
        Output stream.  Default ``sys.stdout``.
    use_colour:
        If False, suppress all ANSI colour codes regardless of terminal type.
        Default True.
    """
    schema = _load_schema(source)
    _print_schema(schema, show_ddl=show_ddl, file=file, use_colour=use_colour)

inspect_fk_path(schema, source, target, *, file=sys.stdout, use_colour=True)

Print all FK and bridge-column chains that connect source to target.

Shows two kinds of paths:

• Direct FK edges (ForeignKeyDef) stored on each table — the declared FOREIGN KEY constraints in the schema.
• Bridge column chains (BridgeColumnDef) computed during schema generation — the transitive multi-hop lookups that ingest uses to propagate FK values through intermediate tables.

Parameters:

Name	Type	Description	Default
schema	SchemaSpec	The SchemaSpec to search.	required
source	str	Table name to start from.	required
target	str	Table name to reach.	required
file	TextIO	Output stream. Default sys.stdout.	stdout
use_colour	bool	If False, suppress all ANSI colour codes regardless of terminal type. Default True.	True

Returns:

Type	Description
bool	True if at least one path was found, False otherwise.

Source code in src/cifflow/inspect/_schema.py

def inspect_fk_path(
    schema: 'SchemaSpec',
    source: str,
    target: str,
    *,
    file: TextIO = sys.stdout,
    use_colour: bool = True,
) -> bool:
    """Print all FK and bridge-column chains that connect *source* to *target*.

    Shows two kinds of paths:

    - **Direct FK edges** (``ForeignKeyDef``) stored on each table — the
      declared ``FOREIGN KEY`` constraints in the schema.
    - **Bridge column chains** (``BridgeColumnDef``) computed during schema
      generation — the transitive multi-hop lookups that ingest uses to
      propagate FK values through intermediate tables.

    Parameters
    ----------
    schema:
        The ``SchemaSpec`` to search.
    source:
        Table name to start from.
    target:
        Table name to reach.
    file:
        Output stream.  Default ``sys.stdout``.
    use_colour:
        If False, suppress all ANSI colour codes regardless of terminal type.
        Default True.

    Returns
    -------
    bool
        ``True`` if at least one path was found, ``False`` otherwise.
    """
    if source not in schema.tables:
        print(c(f'unknown table: {source!r}', RED, file=file, use_colour=use_colour), file=file)
        return False
    if target not in schema.tables:
        print(c(f'unknown table: {target!r}', RED, file=file, use_colour=use_colour), file=file)
        return False

    found = False

    # --- 1. Direct FK chains (BFS over ForeignKeyDef edges) ---
    def _all_fk_paths(src: str, tgt: str) -> list[list[tuple]]:
        """Return all simple FK-edge paths from src to tgt."""
        results: list[list[tuple]] = []
        stack = [(src, [], {src})]
        while stack:
            cur, path, visited = stack.pop()
            for fk in schema.tables[cur].foreign_keys:
                nxt = fk.target_table
                step = (cur, fk.source_columns, nxt, fk.target_columns)
                if nxt == tgt:
                    results.append(path + [step])
                elif nxt not in visited and nxt in schema.tables:
                    stack.append((nxt, path + [step], visited | {nxt}))
        return results

    # Build a lookup: (table, column) → BridgeColumnDef for synthetic columns.
    bridge_by_col: dict[tuple[str, str], 'BridgeColumnDef'] = {
        (bc.table_name, bc.column_name): bc
        for bc in schema.bridge_columns
    }

    def _print_fk_step(from_t: str, src_cols: list[str], tgt_t: str, tgt_cols: list[str], indent: str = '  ') -> None:
        """Print one FK hop, annotating synthetic source columns with their bridge derivation."""
        for src_col, tgt_col in zip(src_cols, tgt_cols):
            bc = bridge_by_col.get((from_t, src_col))
            if bc is not None:
                all_chains = [(bc.hops, bc.bridge_value_column)] + list(bc.fallback_chains)
                n = len(all_chains)
                for i, (chain_hops, chain_val) in enumerate(all_chains):
                    label_str = 'primary' if i == 0 else f'fallback {i}'
                    suffix = f' [{label_str}]' if n > 1 else ''
                    print(
                        f'{indent}{c(src_col, YELLOW, file=file, use_colour=use_colour)}'
                        f' {c(f"(synthetic, derived via bridge{suffix}):", DIM, file=file, use_colour=use_colour)}',
                        file=file,
                    )
                    prev = from_t
                    for via_col, bridge_tbl, bridge_pk in chain_hops:
                        print(
                            f'{indent}  {c(prev, CYAN, file=file, use_colour=use_colour)}.{c(via_col, YELLOW, file=file, use_colour=use_colour)}'
                            f'  ->  {c(bridge_tbl, CYAN, file=file, use_colour=use_colour)}.{c(bridge_pk, GREEN, file=file, use_colour=use_colour)}',
                            file=file,
                        )
                        prev = bridge_tbl
                    print(
                        f'{indent}  {c("value:", DIM, file=file, use_colour=use_colour)} {c(prev, CYAN, file=file, use_colour=use_colour)}.{c(chain_val, GREEN, file=file, use_colour=use_colour)}'
                        f'  =>  {c(from_t, CYAN, file=file, use_colour=use_colour)}.{c(src_col, YELLOW, file=file, use_colour=use_colour)}',
                        file=file,
                    )
            print(
                f'{indent}{c(from_t, CYAN, file=file, use_colour=use_colour)}.{c(src_col, YELLOW, file=file, use_colour=use_colour)}'
                f'  ->  {c(tgt_t, CYAN, file=file, use_colour=use_colour)}.{c(tgt_col, GREEN, file=file, use_colour=use_colour)}',
                file=file,
            )

    fk_paths = _all_fk_paths(source, target)
    if fk_paths:
        found = True
        label = 'FK edge' + ('s' if len(fk_paths) > 1 else '')
        print(c(f'-- {label}: {source}  ->  {target} --', BOLD,
                file=file, use_colour=use_colour), file=file)
        for path in fk_paths:
            for from_t, src_cols, tgt_t, tgt_cols in path:
                _print_fk_step(from_t, src_cols, tgt_t, tgt_cols)
            if len(fk_paths) > 1:
                print(file=file)

    # --- 2. Bridge column chains (BridgeColumnDef) ---
    def _render_chain(hops: list[tuple], val_col: str, from_table: str) -> None:
        prev = from_table
        for via_col, bridge_tbl, bridge_pk in hops:
            print(
                f'    {c(prev, CYAN, file=file, use_colour=use_colour)}.{c(via_col, YELLOW, file=file, use_colour=use_colour)}'
                f'  ->  {c(bridge_tbl, CYAN, file=file, use_colour=use_colour)}.{c(bridge_pk, GREEN, file=file, use_colour=use_colour)}',
                file=file,
            )
            prev = bridge_tbl
        print(
            f'    {c("value:", DIM, file=file, use_colour=use_colour)} {c(prev, CYAN, file=file, use_colour=use_colour)}.{c(val_col, GREEN, file=file, use_colour=use_colour)}',
            file=file,
        )

    bridge_hits = [
        bc for bc in schema.bridge_columns
        if bc.table_name == source and any(bt == target for _, bt, _ in bc.hops)
    ]
    if bridge_hits:
        found = True
        print(c(f'-- bridge columns: {source}  ->  {target} --', BOLD,
                file=file, use_colour=use_colour), file=file)
        for bc in bridge_hits:
            print(
                f'  {c(bc.table_name, CYAN, file=file, use_colour=use_colour)}.{c(bc.column_name, YELLOW, file=file, use_colour=use_colour)}'
                f'  (primary chain):',
                file=file,
            )
            _render_chain(bc.hops, bc.bridge_value_column, bc.table_name)
            for i, (fb_hops, fb_val) in enumerate(bc.fallback_chains, 1):
                print(f'  {c(f"fallback {i}:", DIM, file=file, use_colour=use_colour)}',
                      file=file)
                _render_chain(fb_hops, fb_val, bc.table_name)

    if not found:
        print(
            c(f'no FK or bridge path from {source!r} to {target!r}', RED,
              file=file, use_colour=use_colour),
            file=file,
        )
        partials = [
            pl for pl in schema.partial_links
            if pl.source_table == source and pl.target_table == target
        ]
        if partials:
            print(
                c(f'-- partial links: {source}  ~>  {target} (incomplete FK) --', BOLD,
                  file=file, use_colour=use_colour),
                file=file,
            )
            for pl in partials:
                print(
                    f'  {c(pl.source_table, CYAN, file=file, use_colour=use_colour)}.{c(pl.source_column, YELLOW, file=file, use_colour=use_colour)}'
                    f'  ~>  {c(pl.target_table, CYAN, file=file, use_colour=use_colour)}.{c(pl.target_column, GREEN, file=file, use_colour=use_colour)}',
                    file=file,
                )
                covered = ', '.join(pl.covered_pk_cols) or '(none)'
                missing = ', '.join(pl.missing_pk_cols)
                print(
                    f'    {c("covered PKs:", DIM, file=file, use_colour=use_colour)} {covered}'
                    f'  {c("missing PKs:", DIM, file=file, use_colour=use_colour)} {c(missing, RED, file=file, use_colour=use_colour)}',
                    file=file,
                )
                print(f'    {c("reason:", DIM, file=file, use_colour=use_colour)} {pl.reason}',
                      file=file)

    return found

cifflow.inspect._ingest

inspect_ingest — trace what happens during CIF ingestion.

TraceEvent dataclass

One event captured during :func:inspect_ingest.

Attributes:

Name	Type	Description
kind	str	Category of event. One of: 'warning' — non-fatal semantic issue (e.g. unrecognised tag) 'error' — fatal semantic error
detail	str	Human-readable description of the event.
block_id	Optional[str]	CIF data-block name where the event occurred, if known.
table	Optional[str]	DuckDB table name involved, if applicable.
tag	Optional[str]	CIF tag involved, if applicable.

Source code in src/cifflow/inspect/_ingest.py

@dataclass
class TraceEvent:
    """One event captured during :func:`inspect_ingest`.

    Attributes
    ----------
    kind:
        Category of event.  One of:

        - ``'warning'``      — non-fatal semantic issue (e.g. unrecognised tag)
        - ``'error'``        — fatal semantic error
    detail:
        Human-readable description of the event.
    block_id:
        CIF data-block name where the event occurred, if known.
    table:
        DuckDB table name involved, if applicable.
    tag:
        CIF tag involved, if applicable.
    """

    kind: str
    detail: str
    block_id: Optional[str] = None
    table: Optional[str] = None
    tag: Optional[str] = None

inspect_ingest(cif, db=None, schema=None, *, propagate_fk=False, dataset_id=None, file=None, use_colour=True)

Run ingestion, capture events, and pretty-print a diagnostic trace.

Parameters:

Name	Type	Description	Default
cif	CifFile	Parsed CifFile from build().	required
db	DuckDBPyConnection | None	Open duckdb.DuckDBPyConnection, or None for a fresh in-memory DB.	None
schema	SchemaSpec | None	SchemaSpec used to route tags, or None to route all to fallback.	None
propagate_fk	bool	Forwarded to ingest().	False
dataset_id	str | None	Forwarded to ingest().	None
file	Optional[TextIO]	Where to write the trace. Defaults to sys.stdout.	None
use_colour	bool	If False, suppress all ANSI colour codes regardless of terminal type. Default True.	True

Returns:

Type	Description
list[TraceEvent]	All captured events in occurrence order.

Source code in src/cifflow/inspect/_ingest.py

def inspect_ingest(
    cif: CifFile,
    db: duckdb.DuckDBPyConnection | None = None,
    schema: SchemaSpec | None = None,
    *,
    propagate_fk: bool = False,
    dataset_id: str | None = None,
    file: Optional[TextIO] = None,
    use_colour: bool = True,
) -> list[TraceEvent]:
    """Run ingestion, capture events, and pretty-print a diagnostic trace.

    Parameters
    ----------
    cif:
        Parsed ``CifFile`` from ``build()``.
    db:
        Open ``duckdb.DuckDBPyConnection``, or ``None`` for a fresh in-memory DB.
    schema:
        ``SchemaSpec`` used to route tags, or ``None`` to route all to fallback.
    propagate_fk:
        Forwarded to ``ingest()``.
    dataset_id:
        Forwarded to ``ingest()``.
    file:
        Where to write the trace.  Defaults to ``sys.stdout``.
    use_colour:
        If False, suppress all ANSI colour codes regardless of terminal type.
        Default True.

    Returns
    -------
    list[TraceEvent]
        All captured events in occurrence order.
    """
    if file is None:
        file = sys.stdout

    from cifflow.ingestion.ingest import ingest

    events: list[TraceEvent] = []

    print(c('-- inspect_ingest --', BOLD, DIM, file=file, use_colour=use_colour), file=file)

    try:
        _, ingest_errors = ingest(
            cif, db, schema=schema,
            propagate_fk=propagate_fk,
            dataset_id=dataset_id,
        )
        for msg in ingest_errors:
            events.append(TraceEvent(kind='warning', detail=msg))

    except ValueError as exc:
        events.append(TraceEvent(kind='error', detail=str(exc)))

    except Exception as exc:
        events.append(TraceEvent(kind='error', detail=str(exc)))

    warnings_ev = [e for e in events if e.kind == 'warning']
    errors_ev = [e for e in events if e.kind == 'error']

    if warnings_ev:
        print(c(f'  {len(warnings_ev)} semantic warning(s):', YELLOW,
                file=file, use_colour=use_colour), file=file)
        for ev in warnings_ev:
            print(f'    {c("~", YELLOW, file=file, use_colour=use_colour)}  {ev.detail}',
                  file=file)

    if errors_ev:
        print(c(f'  {len(errors_ev)} error(s):', RED, BOLD,
                file=file, use_colour=use_colour), file=file)
        for ev in errors_ev:
            print(f'    {c("!", RED, file=file, use_colour=use_colour)}  {ev.detail}',
                  file=file)

    if not warnings_ev and not errors_ev:
        print(c('  Ingestion completed with no warnings.', GREEN,
                file=file, use_colour=use_colour), file=file)

    _print_trace_summary(events, file, use_colour=use_colour)
    return events