kg_microbe.merge_utils package

Submodules

kg_microbe.merge_utils.artifact_manifest module

Portable manifests for the exact cleaned KGX TSV bytes shipped by a merge.

kg_microbe.merge_utils.artifact_manifest.build_provenance(config_file, repo_root, ignore=())

Record config identity and repository inputs without depending on external stats.

Return type:

Dict

kg_microbe.merge_utils.artifact_manifest.write_graph_archive(archive, files, provenance=None)

Atomically ship TSVs and their byte-accurate manifest; retain the old archive on failure.

Return type:

None

kg_microbe.merge_utils.artifact_manifest.write_loose_manifest(destination, files, provenance=None)

Describe the matching uncompressed TSV pair using paths relative to the manifest.

Return type:

None

kg_microbe.merge_utils.external_node_closure module

Evidence-backed closure for referenced external nodes missing from source declarations.

Only anonymous endpoints are considered. Retired identifiers use explicit authority replacements; unsupported obsolete assertions follow the ontology transform’s exclusion policy and are retained verbatim in a disposition TSV. No name matching, accession-prefix substitution, or taxonomy inference occurs.

kg_microbe.merge_utils.external_node_closure.resolve_external_reference_bundle(node_paths, edge_paths, raw_dir, report_path, *, discover_references=False, include_go=True, strict=False, quoting=0, known_declarations=frozenset({}), consumed_authorities=None)

Resolve one staged source bundle, preserving file membership and every original assertion.

The single-pair compatibility API retains its existing anonymous-node behavior. Source finalization additionally discovers referenced external IDs not declared locally: KGX must no longer invent those nodes before authority normalization can see them. GO is handled by its all-ID source authority normalizer when include_go is false. No adapters are created in pool workers; callers invoke this after producer workers have completed.

kg_microbe.merge_utils.external_node_closure.resolve_external_references(nodes_path, edges_path, raw_dir, report_path)

Normalize anonymous external references using exact local proof and retain full evidence.

Return disposition counts. Files are replaced atomically only after all authority reads and output writes succeed; failures propagate to prevent publishing an apparently successful, partially repaired archive. The compatibility caller supplies unpublished staging files. The production pipeline invokes the bundle API during source finalization, not merge.

kg_microbe.merge_utils.invariants module

Invariants checked against the merged graph, where sources actually meet.

Each transform can only police the edges it writes itself. kgmicrobe.strain:* is a deliberately shared namespace – LPSN mints the same CURIEs BacDive does so the merge reconciles them – so a rule about those nodes is only really enforced after the merge. #892 fixed multi-parent strains inside the BacDive transform; this module is the check that notices if another source reintroduces them (#896).

kg_microbe.merge_utils.invariants.EXPECTED_STUB_PATTERNS = ((re.compile('GTDB:[dpcofgs]__.+'), 'GTDB taxon strings, referenced as cross-references'), (re.compile('gold:G[spb]\\d+'), 'GOLD study, project and biosample identifiers'))

Narrow exceptions for identifiers we deliberately do not declare. Keep the namespace canonical: uppercase GOLD is always a legacy-namespace defect.

kg_microbe.merge_utils.invariants.EXPECTED_STUB_PREFIXES = {'IMG': 'IMG genome and taxon identifiers, referenced from GOLD'}

Only prefixes whose every identifier is an external cross-reference belong here. GOLD organisms and GTDB assemblies have owner transforms, so exempting their entire prefixes hid namespace defects affecting 31 K nodes (#1050/1051).

kg_microbe.merge_utils.invariants.NAMED_THING_CATEGORY = 'biolink:NamedThing'

What KGX types a node it invented for an undeclared endpoint.

kg_microbe.merge_utils.invariants.STRAIN_PARENT_REPORT = 'merged_strain_parent_violations.tsv'

Written beside merged-kg_nodes.tsv / merged-kg_edges.tsv.

kg_microbe.merge_utils.invariants.STUB_NODE_REPORT = 'merged_stub_nodes.tsv'

Written beside the merged TSVs, alongside the strain-parent report.

kg_microbe.merge_utils.invariants.check_merged_invariants(edges_file, output_dir=None, nodes_file=None)

Run the merged-graph invariants and write their report.

The report is written whether or not anything was found: an absent file would be indistinguishable from a clean run, and a clean run is the thing worth being able to demonstrate.

Parameters:

  • edges_file (Path) – Path to merged-kg_edges.tsv.

  • output_dir (Optional[Path]) – Where to write the report; defaults to the edge file’s directory.

  • nodes_file (Optional[Path]) – Path to the matching nodes TSV. Defaults to deriving it from the edge file’s name, for callers pointing at a bare edge dump – but the merge hands over the path it already built, because a derivation that fails looks exactly like a graph with no stubs (#936).

Return type:

int

Returns:

Number of violating strain nodes.

kg_microbe.merge_utils.invariants.expected_stub_reason(curie)

Return the justification for an intentionally undeclared CURIE, if any.

Return type:

str

kg_microbe.merge_utils.invariants.find_multi_parent_strains(edges_file)

Return strain nodes carrying more than one NCBITaxon parent in the merged graph.

Streams the edge file rather than loading it: the merged graph is tens of millions of rows, and only the strain rows are retained.

Parameters:

edges_file (Path) – Path to merged-kg_edges.tsv.

Return type:

Dict[str, Tuple[Set[str], Set[str]]]

Returns:

{strain CURIE: (parent CURIEs, knowledge sources)} for every strain with more than one distinct parent. Empty when the invariant holds.

kg_microbe.merge_utils.invariants.find_stub_nodes(nodes_file)

Return nodes KGX synthesized because an edge referenced something undeclared.

The merge leaves no endpoint without a node row – KGX invents one – so the question worth asking after a merge is not “what is missing” but “what arrived as an invention”. Those rows are typed biolink:NamedThing and carry no name, which is the shape #892 objected to.

Parameters:

nodes_file (Path) – Path to merged-kg_nodes.tsv.

Return type:

Dict[str, List[str]]

Returns:

{CURIE prefix: [CURIEs]}, sorted, empty when none.

kg_microbe.merge_utils.invariants.strain_parent_rows(violations)

Render violations as report rows, sorted so a diff between runs is readable.

Parameters:

violations (Dict[str, Tuple[Set[str], Set[str]]]) – Result of find_multi_parent_strains().

Return type:

List[List]

Returns:

Rows matching STRAIN_PARENT_REPORT_HEADER.

kg_microbe.merge_utils.invariants.stub_incident_rows(edges_file, stubs)

Attribute every anonymous node to its incident assertions, not just prefix samples.

Only stub IDs, counts and distinct predicate/source strings are retained in memory. Source cells are recorded verbatim: malformed legacy serialization must remain inspectable, not be silently interpreted as Python (#1070). A self-loop counts once for its one incident node. Duplicate headers use the first occurrence, consistent with the other merge invariants (#915).

Return type:

List[List]

kg_microbe.merge_utils.invariants.stub_node_rows(stubs)

Render stub counts per prefix and expectation, flagging unintended ones.

A mixed prefix produces separate yes/no rows: 25 taxon references must not cause 15,584 mis-prefixed assembly references to be called expected (#1050).

Parameters:

stubs (Dict[str, List[str]]) – Result of find_stub_nodes().

Return type:

List[List]

Returns:

Rows matching STUB_NODE_REPORT_HEADER.

kg_microbe.merge_utils.kgx_source module

kg_microbe.merge_utils.local_context module

Scope KGX conversion/merge prefix contexts to the pinned local model, including spawned workers.

kg_microbe.merge_utils.local_context.local_prefix_context()

Seed all real KGX prefix readers, restoring exact prior entries after nested users finish.

kg_microbe.merge_utils.merge_kg module

Merging module.

kg_microbe.merge_utils.merge_kg.load_and_merge(yaml_file, processes=1, sources=None)

Load and merge sources defined in the config YAML.

sources restricts the merge to a subset of the config’s sources. KGX holds every source’s graph in the parent process at once — kgx.cli.cli_utils.merge collects them all before calling merge_all_graphs — so peak memory scales with the whole set, not with the largest member. Merging in stages is the only way to keep a very large source (PREGO: 44.7M edges) from coexisting with all the others.

Parameters:

  • yaml_file (str) – A string pointing to a KGX compatible config YAML.

  • processes (int) – Number of processes to use. Each concurrent process holds its own source graph, so raising this raises peak memory.

  • sources (Optional[List[str]]) – Optional subset of source keys to merge. None merges all.

Return type:

MultiDiGraph

Returns:

The pre-serialization KGX in-memory graph. Serialization projection affects the published TSV/archive, not this returned object; use its manifest for final byte identity and row counts. No post-merge semantic repairs are permitted. Sources must be explicitly finalized first.

Raises:

  • KeyError – If a requested source is absent from the config.

  • ValueError – If any release destination is unsupported, malformed, or colliding.

kg_microbe.merge_utils.merge_kg.merge(*args, **kwargs)

Invoke KGX with local schema and provenance-preserving graph export (#1049).

kg_microbe.merge_utils.merge_kg.parse_load_config(yaml_file)

Parse load config YAML.

Parameters:

yaml_file (str) – A string pointing to a KGX compatible config YAML.

Return type:

Dict

Returns:

Dict: The config as a dictionary.

kg_microbe.merge_utils.source_admission module

Bind one merge to unchanged admitted files without copying graph-scale inputs (#1100).

class kg_microbe.merge_utils.source_admission.SourceAdmission

Bases: object

Retain exact checked byte identities plus conservative filesystem change observations.

bind_path(path)

Keep a lexical input locator tied to its original resolved target, including symlinks.

capture(path, *, retain=False, optional=False)

Observe bytes before the caller validates them, caching graph hashes but not whole graphs.

capture_resolution(resolver, *args)

Bind precedence-sensitive input resolution, not just the originally chosen lexical path.

capture_tree(path)

Bind Python package membership too, so newly added producer helpers cannot evade checks.

read_json(path)

Parse exactly the metadata bytes whose hash is retained in the admission set.

verify(*, metadata_only=False)

Reject drift from original admission; this is not a substitute-current-record freshness check.

kg_microbe.merge_utils.source_freshness module

Enforce recorded producer/dependency freshness before KGX without changing transform contracts.

kg_microbe.merge_utils.source_freshness.verify_source_freshness(paths, *, admission=None)

Reject stale selected sources and declared dependencies before staging or running KGX.

kg_microbe.merge_utils.stats_provenance module

Make merged_graph_stats.yaml say what produced it, and count every predicate.

KGX’s generate_graph_stats writes only what it computes, and it resolves predicates through the Biolink toolkit: anything it cannot resolve is recorded as None. METPO predicates are deliberately not Biolink, so two thirds of the graph’s edges were invisible in the stats file the README points readers at (#993). The file also carried no merge date, commit, config or input digests, so a stale copy could not be told from a current one except by diffing counts against a graph one already trusted (#1013).

After the merge writes the stats, annotate_graph_stats() adds two blocks the merge step already knows how to fill:

provenance

when, from which commit (short HEAD, -dirty if tracked files other than the stats file differ), which merge config, which edges artifact (archive and member, or loose file), and each source’s source_fingerprint.json digest or no marker.

edge_stats.count_by_raw_predicate

the predicate column counted directly from the merged edges TSV, no resolution, so METPO:* predicates appear next to biolink:* and the counts sum to the edge total.

Idempotent: both blocks are replaced, never appended, so re-annotating a file yields the same shape.

kg_microbe.merge_utils.stats_provenance.annotate_graph_stats(stats_file, edges_file, yaml_file, repo_root, now=None, edges_archive=None, published_edges_file=None, finalized_nodes_file=None)

Add provenance and edge_stats.count_by_raw_predicate to a stats file.

Parameters:

  • stats_file (Path) – The YAML KGX wrote.

  • edges_file (Path) – The merged edges TSV (loose, not the archive).

  • yaml_file (Path) – The merge config that produced both.

  • repo_root (Path) – Checkout root, for git describe and the markers.

  • now (Optional[datetime]) – Timestamp to record; defaults to UTC now.

  • edges_archive (Optional[Path]) – Published archive containing the edges; when supplied, record it and its member instead of the temporary loose TSV (#1055).

  • published_edges_file (Optional[Path]) – Final loose locator, while counts read edges_file in staging.

  • finalized_nodes_file (Optional[Path]) – Opt into a full streaming recount of the finalized TSV pair; preserve the entire original KGX summary only under pre_normalization_stats.

Return type:

Dict

Returns:

The annotated stats dict, as written.

Raises:

ValueError – If the raw predicate total disagrees with KGX’s total_edges – the two counted different files.

kg_microbe.merge_utils.stats_provenance.count_raw_predicates(edges_file)

Count the predicate column of a KGX edges TSV as written, no resolution.

Return type:

Counter

kg_microbe.merge_utils.stats_provenance.git_commit(repo_root, ignore=())

Return the short HEAD commit, with -dirty when tracked files differ.

git describe --dirty cannot be used as-is: at annotation time KGX has just rewritten the stats file, so the tree is always dirty by exactly that file. Paths in ignore (the stats file) are excluded from the dirtiness check; untracked files never count. unknown when git or the checkout is unavailable.

Return type:

str

kg_microbe.merge_utils.stats_provenance.source_markers(config, repo_root)

Map each merge source to its transform’s fingerprint digest.

The source’s input files name the transform directory (data/transformed/<source>/...); its source_fingerprint.json is the marker the freshness check reads. no marker is recorded rather than omitted, so an unfingerprinted input is visible in the artifact.

Return type:

Dict[str, str]

kg_microbe.merge_utils.stats_provenance.stats_filename_from_config(config)

Return the stats filename the merge config asks KGX to write, if any.

Return type:

Optional[str]

Module contents

Merge module.

kg_microbe.merge_utils.load_and_merge(yaml_file, processes=1, sources=None)

Load and merge sources defined in the config YAML.

sources restricts the merge to a subset of the config’s sources. KGX holds every source’s graph in the parent process at once — kgx.cli.cli_utils.merge collects them all before calling merge_all_graphs — so peak memory scales with the whole set, not with the largest member. Merging in stages is the only way to keep a very large source (PREGO: 44.7M edges) from coexisting with all the others.

Parameters:

  • yaml_file (str) – A string pointing to a KGX compatible config YAML.

  • processes (int) – Number of processes to use. Each concurrent process holds its own source graph, so raising this raises peak memory.

  • sources (Optional[List[str]]) – Optional subset of source keys to merge. None merges all.

Return type:

MultiDiGraph

Returns:

The pre-serialization KGX in-memory graph. Serialization projection affects the published TSV/archive, not this returned object; use its manifest for final byte identity and row counts. No post-merge semantic repairs are permitted. Sources must be explicitly finalized first.

Raises:

  • KeyError – If a requested source is absent from the config.

  • ValueError – If any release destination is unsupported, malformed, or colliding.

kg_microbe.merge_utils.parse_load_config(yaml_file)

Parse load config YAML.

Parameters:

yaml_file (str) – A string pointing to a KGX compatible config YAML.

Return type:

Dict

Returns:

Dict: The config as a dictionary.