avoid republishing

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pubmate"
-version = "0.2.4"
+version = "0.2.5"
 description = "Helper tool aiding with publishing nanopublications departing from a linkml schema"
 readme = "README.md"
 requires-python = ">=3.12"

src/pubmate/cli/mint.py

@@ -4,6 +4,7 @@
 import yaml
 
 from pubmate import IdentifierGenerator
+from pubmate.mint import ID_METHODS
 
 # Configure logging
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
@@ -17,7 +18,7 @@
 @click.option("--target", "-t", "target_name", required=True, help="Name of the target entity list in the data file.")
 @click.option("--namespace", required=True, help="Namespace prefix used to create identifiers.")
 @click.option("--id-key", default="id", help="Field name containing the identifier.")
-@click.option("--method", default="ulid", type=click.Choice(["ulid", "uuid", "hash"]), help="ID generation method.")
+@click.option("--method", default="ulid", type=click.Choice(ID_METHODS), help="ID generation/checking method.")
 @click.option(
     "--output", "output_path", default=None, help="Optional output path. Defaults to overwriting the input file."
 )
@@ -69,10 +70,11 @@ def cli(
             if not label_value:
                 raise ValueError(f"Entity at index {i} has no '{preflabel}' value: {entity}")
 
-            # Keep existing valid URIs unless forced. They may have been minted with
+            # Keep existing IDs that match the selected minting pattern. Keep
+            # other valid URIs too, unless forced: they may have been minted with
             # another strategy, another namespace, or outside PubMate entirely.
             if current_id and not force:
-                if id_generator.is_valid_uri(current_id):
+                if id_generator.is_valid_id(current_id, method=method) or id_generator.is_valid_uri(current_id):
                     id_generator.register_id(current_id)
                     logger.debug(f"Keeping existing ID: {current_id}")
                     continue

src/pubmate/incremental.py

@@ -47,11 +47,11 @@
 
 import rdflib
 
-from pubmate._nanopub_build import preferred_label
+from pubmate._nanopub_build import UNSET as _UNSET, preferred_label
 from pubmate.fingerprint import fingerprint_term
 from pubmate.idmap import IdMap, IdMapEntry
 from pubmate.migrate import MintedSupersession
-from pubmate.minting import MintBatch, SequentialMinter, TermInput
+from pubmate.minting import MintBatch, MintedTerm, SequentialMinter, TermInput, is_trusty_thing_uri
 from pubmate.rdf2nanopub import sign_and_publish
 from pubmate.references import resolve_references
 from pubmate.supersede import SupersessionBuilder
@@ -99,12 +99,47 @@ def publish_incremental(
     default_suggester = minter.default_suggester_orcid
 
     for term in terms:
-        current_fp = fingerprint_term(term, minter.builder, default_suggester=default_suggester)
-
         entry = result.id_map.resolve(term.term_id)
+        already_trusty = is_trusty_thing_uri(term.term_id, namespace=minter.builder.namespace)
+        fingerprint_introduces = (
+            term.term_id
+            if (entry is None and already_trusty)
+            or (
+                entry is not None
+                and entry.old_id == entry.thing_uri
+                and is_trusty_thing_uri(entry.thing_uri, namespace=minter.builder.namespace)
+            )
+            else _UNSET
+        )
+        current_fp = fingerprint_term(
+            term,
+            minter.builder,
+            default_suggester=default_suggester,
+            introduces=fingerprint_introduces,
+        )
 
         if entry is None:
-            minted = minter.mint(term, dry_run=dry_run)
+            if already_trusty:
+                fixed = rdflib.URIRef(term.term_id)
+                full = resolve_references(
+                    term.assertion,
+                    namespace=minter.builder.namespace,
+                    subject=placeholder,
+                    new_subject=fixed,
+                    thing_uris=result.id_map.resolution_map,
+                )
+                np = minter.builder.build(
+                    full,
+                    suggester_orcid=term.suggester_orcid or default_suggester,
+                    label=term.label or preferred_label(full, fixed),
+                    derived_from=term.derived_from,
+                    introduces=term.term_id,
+                )
+                np_uri = sign_and_publish(np, dry_run=dry_run)
+                minted = MintedTerm(term_id=term.term_id, thing_uri=term.term_id, np_uri=np_uri, nanopub=np)
+                logger.info("Published already-minted term %s -> %s", term.term_id, np_uri)
+            else:
+                minted = minter.mint(term, dry_run=dry_run)
             result.id_map.add(
                 IdMapEntry(term.term_id, minted.thing_uri, minted.np_uri, current_fp),
                 overwrite=True,

src/pubmate/mint.py

@@ -1,8 +1,8 @@
+import json
 import hashlib
 import logging
 import os
 import re
-import json
 
 from typing import Optional, Set
 from urllib.parse import urlsplit
@@ -13,6 +13,16 @@
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
 
+ID_METHODS = ("ulid", "uuid", "hash", "trusty")
+GENERATIVE_ID_METHODS = ("ulid", "uuid", "hash")
+_UNIQUE_PART_PATTERNS = {
+    "uuid": r"[0-9a-f]{8}",
+    "hash": r"[0-9a-f]{10}",
+    "ulid": r"[0-9A-HJKMNP-TV-Z]{26}",
+    # Trusty artifact code, e.g. "RA" followed by a base64url-ish hash.
+    "trusty": r"RA[A-Za-z0-9_\-]{40,}",
+}
+
 
 class IdentifierGenerator:
     def __init__(
@@ -49,38 +59,24 @@ def is_valid_uri(self, key: str) -> bool:
 
         return True
 
-    def is_valid_id(self, key: str, method: str = "ulid") -> bool:
-        # Check if key starts with namespace
-        if not key.startswith(self.namespace):
-            return False
-
-        # Remove namespace to get the remaining part
-        remaining = key[len(self.namespace) :]
+    def id_pattern(self, method: str = "ulid") -> re.Pattern:
+        """Return the full identifier regex for ``method`` in this namespace."""
+        try:
+            unique_pattern = _UNIQUE_PART_PATTERNS[method]
+        except KeyError:
+            raise NotImplementedError(f"Unknown method: {method}. Available methods: {', '.join(ID_METHODS)}") from None
 
-        # Define expected unique part patterns based on method
-        if method == "uuid":
-            unique_pattern = r"^[0-9a-f]{8}$"  # 8 hex chars
-        elif method == "hash":
-            unique_pattern = r"^[0-9a-f]{10}$"  # 10 hex chars
-        elif method == "ulid":
-            unique_pattern = r"^[0-9A-HJKMNP-TV-Z]{26}$"  # full ULID
+        if self.type_prefix is None:
+            suffix_pattern = unique_pattern
         else:
-            raise NotImplementedError
-
-        # Check pattern based on whether type_prefix is used
-        if self.type_prefix is not None:
-            # Expected format: {type_prefix}-{unique_part}
-            expected_prefix = f"{self.type_prefix}-"
-            if not remaining.startswith(expected_prefix):
-                return False
-            # Extract unique part after type_prefix and dash
-            unique_part = remaining[len(expected_prefix) :]
-        else:
-            # Expected format: {unique_part} directly
-            unique_part = remaining
+            suffix_pattern = f"{re.escape(self.type_prefix)}-{unique_pattern}"
+
+        return re.compile(f"^{re.escape(self.namespace)}{suffix_pattern}$")
 
-        # Validate the unique part matches the expected pattern
-        return bool(re.match(unique_pattern, unique_part))
+    def is_valid_id(self, key: str, method: str = "ulid") -> bool:
+        if not isinstance(key, str):
+            return False
+        return bool(self.id_pattern(method).match(key))
 
     def register_id(self, identifier: str) -> None:
         self.registered_ids.add(identifier)
@@ -134,8 +130,13 @@ def generate_id(
                 hash_str = self.hash_dict(salted_entity)
                 unique_part = hash_str[:10]
 
+            elif method == "trusty":
+                raise ValueError(
+                    "Cannot generate a trusty artifact-code identifier in pubmate-mint; "
+                    "publish/sign a nanopublication instead."
+                )
             else:
-                raise ValueError(f"Unknown method: {method}. Available methods: ulid, uuid, hash")
+                raise ValueError(f"Unknown method: {method}. Available methods: {', '.join(GENERATIVE_ID_METHODS)}")
 
             # Construct the full identifier
             if self.type_prefix is None:

src/pubmate/minting.py

@@ -153,6 +153,11 @@ def _artifact_code(np_uri: str) -> str:
     return match.group(0)
 
 
+def is_trusty_thing_uri(uri: str, *, namespace: str) -> bool:
+    """Return whether ``uri`` is a namespace-local trusty artifact-code identifier."""
+    return bool(re.match(f"^{re.escape(namespace)}{_ARTIFACT_CODE_RE.pattern}$", uri))
+
+
 class SequentialMinter:
     """Mint defining nanopubs one by one with a configured builder.
 

tests/test_cli_mint.py

@@ -65,3 +65,57 @@ def test_mint_cli_keeps_valid_uri_that_does_not_match_selected_method(tmp_path)
 
     rendered = yaml.safe_load(result.stdout)
     assert rendered["vocabulary_terms"][0]["id"] == existing_id
+
+
+def test_mint_cli_keeps_trusty_id_when_method_is_trusty(tmp_path) -> None:
+    data_path = tmp_path / "terms.yaml"
+    existing_id = "https://example.org/terms/RA" + ("a" * 41)
+    data_path.write_text(
+        ("vocabulary_terms:\n" "  - name: Alpha\n" f"    id: {existing_id}\n"),
+        encoding="utf-8",
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(
+        cli,
+        [
+            "--data",
+            str(data_path),
+            "--target",
+            "vocabulary_terms",
+            "--namespace",
+            "https://example.org/terms/",
+            "--method",
+            "trusty",
+            "--dry-run",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+
+    rendered = yaml.safe_load(result.stdout)
+    assert rendered["vocabulary_terms"][0]["id"] == existing_id
+
+
+def test_mint_cli_cannot_generate_missing_trusty_id(tmp_path, caplog) -> None:
+    data_path = tmp_path / "terms.yaml"
+    data_path.write_text("vocabulary_terms:\n  - name: Alpha\n", encoding="utf-8")
+
+    runner = CliRunner()
+    result = runner.invoke(
+        cli,
+        [
+            "--data",
+            str(data_path),
+            "--target",
+            "vocabulary_terms",
+            "--namespace",
+            "https://example.org/terms/",
+            "--method",
+            "trusty",
+            "--dry-run",
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "Cannot generate a trusty artifact-code identifier" in caplog.text

tests/test_cli_mint_publish.py

@@ -73,6 +73,35 @@ def test_mint_publish_dry_run_writes_trig_and_idmap(tmp_path) -> None:
     assert entry.np_uri.startswith("https://w3id.org/np/RA")
 
 
+def test_mint_publish_keeps_existing_trusty_subject_without_idmap(tmp_path) -> None:
+    trusty_id = f"{NAMESPACE}RA" + ("a" * 41)
+    assertions = tmp_path / "assertions"
+    assertions.mkdir()
+    g = rdflib.Graph()
+    s = rdflib.URIRef(trusty_id)
+    g.add((s, RDF.type, SKOS.Concept))
+    g.add((s, RDFS.label, rdflib.Literal("Caffeine")))
+    g.serialize(destination=assertions / "caffeine.ttl", format="turtle")
+    out = tmp_path / "published"
+    idmap = tmp_path / "id-map.tsv"
+
+    result = CliRunner().invoke(
+        cli,
+        ["-a", str(assertions), "--output-dir", str(out), "--id-map-file", str(idmap), "--dry-run"],
+    )
+
+    assert result.exit_code == 0, result.output
+    parsed = IdMap.from_tsv(idmap.read_text(encoding="utf-8"))
+    entry = parsed[trusty_id]
+    assert entry.thing_uri == trusty_id
+    assert entry.np_uri.startswith("https://w3id.org/np/RA")
+
+    np = rdflib.Dataset()
+    np.parse(next(out.glob("*.trig")), format="trig")
+    subjects = {str(s) for s, _p, _o, _g in np.quads((None, None, None, None))}
+    assert trusty_id in subjects
+
+
 def test_mint_publish_adds_part_of_type_and_template(tmp_path) -> None:
     DCTERMS = rdflib.Namespace("http://purl.org/dc/terms/")
     NPX = rdflib.Namespace("http://purl.org/nanopub/x/")

tests/test_incremental.py

@@ -62,6 +62,37 @@ def test_new_term_is_minted_and_records_fingerprint():
     assert entry.fingerprint != ""
 
 
+def test_new_trusty_term_is_published_without_reminting_thing_uri():
+    thing = f"{NAMESPACE}RA" + ("a" * 41)
+
+    result = _run([_term(thing, "Alpha")])
+
+    assert result.superseded == []
+    assert [m.term_id for m in result.minted.terms] == [thing]
+    minted = result.minted.terms[0]
+    assert minted.thing_uri == thing
+    assert minted.np_uri.startswith("https://w3id.org/np/RA")
+    assert rdflib.URIRef(thing) in set(minted.nanopub.assertion.subjects())
+    assert rdflib.URIRef(f"{NAMESPACE}RA") not in set(minted.nanopub.assertion.subjects())
+
+    entry = result.id_map[thing]
+    assert entry.thing_uri == thing
+    assert entry.np_uri == minted.np_uri
+    assert entry.fingerprint != ""
+
+
+def test_new_trusty_term_rerun_is_skipped():
+    thing = f"{NAMESPACE}RA" + ("a" * 41)
+    first = _run([_term(thing, "Alpha")])
+
+    again = _run([_term(thing, "Alpha")], existing=first.id_map)
+
+    assert again.minted.terms == []
+    assert again.superseded == []
+    assert again.skipped == [thing]
+    assert again.id_map[thing] == first.id_map[thing]
+
+
 def test_unchanged_term_is_skipped():
     first = _run([_term("alpha", "Alpha")])
     again = _run([_term("alpha", "Alpha")], existing=first.id_map)

tests/test_mint.py

@@ -32,6 +32,23 @@ def test_is_valid_uri_accepts_absolute_uris_without_requiring_minting_pattern()
     assert not generator.is_valid_id("https://example.org/terms/alpha", method="ulid")
 
 
+def test_is_valid_id_accepts_trusty_artifact_code_in_namespace() -> None:
+    generator = IdentifierGenerator(namespace="https://example.org/terms/")
+    artifact = "RA" + ("a" * 41)
+
+    assert generator.is_valid_id(f"https://example.org/terms/{artifact}", method="trusty")
+    assert not generator.is_valid_id(f"https://other.example/terms/{artifact}", method="trusty")
+    assert not generator.is_valid_id(f"https://example.org/terms/{artifact}/extra", method="trusty")
+
+
+def test_is_valid_id_accepts_type_prefixed_trusty_artifact_code() -> None:
+    generator = IdentifierGenerator(namespace="https://example.org/terms/", type_prefix="chem")
+    artifact = "RA" + ("a" * 41)
+
+    assert generator.is_valid_id(f"https://example.org/terms/chem-{artifact}", method="trusty")
+    assert not generator.is_valid_id(f"https://example.org/terms/{artifact}", method="trusty")
+
+
 def test_is_valid_uri_rejects_relative_or_malformed_values() -> None:
     generator = IdentifierGenerator(namespace="https://example.org/terms/")
 
@@ -45,3 +62,10 @@ def test_generate_id_raises_runtimeerror_when_attempts_exhausted() -> None:
 
     with pytest.raises(RuntimeError, match="Could not generate a unique identifier for 'Alpha' after 0 attempts"):
         generator.generate_id({"name": "Alpha"}, method="hash", max_attempts=0)
+
+
+def test_generate_id_rejects_trusty_method() -> None:
+    generator = IdentifierGenerator(namespace="https://example.org/")
+
+    with pytest.raises(ValueError, match="Cannot generate a trusty artifact-code identifier"):
+        generator.generate_id({"name": "Alpha"}, method="trusty")