Skip to content

stacs.register

Register a published catalogue into pgstac, and prove it arrived.

Upsert only, collection before items, and nothing in this module deletes. pgstac.items.collection is ON DELETE CASCADE: dropping a collection row destroys every item in it, which is how a delete-then-load once took a catalogue offline. A load that fails here leaves the previous rows serving: an upsert never removes one.

The one exception is pgstac's, not ours: a queryables load replaces every row of the name that the collection is served, including a global or shared one. It is made only for a queryable served differently from its declaration, never for one already served as declared, and another collection's queryable it changed fails the run.

Writes go over ssh to the STAC host, which runs pypgstac load there: the database password never leaves the host, and a STAC API without the transaction extension cannot be written through anyway. Each load first asks the host's connection which database it reached, and refuses anything but the one named. The host, database, environment file and launcher are all the caller's (Transport); nothing here names a deployment.

Modes:

  • verify: compare every published body with what the API serves and report missing, changed and orphaned items and the collection's state. Writes nothing.
  • drift: register what the API is missing or serves with a different body. Stateless: it asks the API what it has, so a skipped run is picked up by the next one.
  • all: register every published item.
  • ids: register exactly the ids in a file.

RegisterError

Bases: Exception

A refusal. Raised before anything is written wherever the check allows.

Transport dataclass

How to reach pgstac: ssh to host, then pypgstac load there.

Every value is the caller's. password_env is the NAME of a variable the host's environment (usually env_file) defines, never a password: the remote script is an ssh argument, and argv is visible in ps on both ends.

Target dataclass

One catalogue: where it is published, where it is served, how it is written, and the asset rules and queryables it declares for itself.

check()

The local half of "refuse before writing": types and ranges, so a bad chunk cannot first fail in the read-back AFTER the upsert. Whether the API answers is checked separately, by _api_probe, before any write.

ndjson_write(paths, out, expect_collection=None)

Compact one item JSON per line. Returns the number of lines written.

json.dumps never emits a raw newline, so a record cannot straddle lines however odd the source formatting is.

expect_collection is the last checkpoint before pgstac. pgstac routes each item by its OWN collection field, so an item whose body still names the previous collection upserts into the previous collection SUCCESSFULLY, with no error anywhere.

Examples:

>>> import json, os, tempfile
>>> tmp = tempfile.TemporaryDirectory()
>>> item = os.path.join(tmp.name, "a.json")
>>> with open(item, "w") as f:
...     json.dump({"id": "a", "collection": "old-collection"}, f, indent=2)
>>> out = os.path.join(tmp.name, "items.ndjson")
>>> ndjson_write([item], out)
1
>>> open(out).read()
'{"id":"a","collection":"old-collection"}\n'
>>> ndjson_write([item], out, expect_collection="my-collection")
Traceback (most recent call last):
...
stacs.register.RegisterError: ...a.json names collection 'old-collection', expected
'my-collection'. Loading it would register the item into 'old-collection' without
erroring.
>>> tmp.cleanup()

remote_script(t, kind, expected, collection_id=None)

The script the host runs, payload on stdin. Every caller value is quoted.

kind is items, collections or queryables. Queryables are loaded with load_queryables for collection_id alone, and never with --delete_missing: pypgstac replaces every row of a name that overlaps the collections it is given, so naming any other collection would narrow or remove that collection's queryable.

The line count is checked on the RECEIVING side: if the sender dies mid-stream the remote cat sees EOF, and pypgstac would load a syntactically valid short file and report success.

The caller's environment file is the caller's code, so it runs in a SUBSHELL with the load, never in the shell that holds this script's state. It cannot turn -e off for the parent, nor reassign the payload path (readonly, and namespaced). The success line STACS_LOADED <n> <kind> <db> is printed inside that subshell only after pypgstac returns 0, so neither an exit 0 nor set +e in the environment file can produce it.

The PG* exports come AFTER the environment file is sourced, so the database validated here is the one asked for, whatever the file sets. Before the load, the launcher is asked which database its connection reaches (current_database()), and anything but t.db is refused before pypgstac writes: a launcher that does not pass PGDATABASE on connects elsewhere, and a read-back cannot see it where that database already serves the same bodies. The name read is the <db> of the success line.

Atomicity: a collection load is one row. An items load is NOT one transaction -- pypgstac commits each chunk per partition -- so a failed items load can leave some of its items committed. Every load is an upsert, so re-running converges.

Examples:

>>> t = Transport(host="user@stac.example.invalid", db="pgstac",
...               env_file="/srv/stac/.env")
>>> script = remote_script(t, "items", 2)
>>> [line.strip() for line in script.splitlines()
...  if line.strip().startswith(("pypgstac", "__stacs_db="))]
["__stacs_db=$(pypgstac _db query_one 'SELECT current_database()') || exit 1",
 'pypgstac load items "$__stacs_tmp" --method upsert || exit 1']
>>> script.splitlines()[-1]
'__stacs_ok=1'
>>> import shlex
>>> script = remote_script(t, "queryables", 1, collection_id="my-collection")
>>> line = next(l for l in script.splitlines() if "load_queryables" in l)
>>> shlex.split(line)[1:4]
['load_queryables', '$__stacs_tmp', "--collection_ids=['my-collection']"]

probe(t)

Refuse before the expensive stage if the host cannot be reached.

load(t, kind, paths, expect_collection=None, dryrun=False, log=print)

Upsert JSON files into pgstac on the host. Returns the number loaded.

paths is a list, never argv: 102k filenames is ~6 MB against a ~2 MB ARG_MAX. The payload is assembled completely to a local file before ssh starts, so a refusal half way through (expect_collection) sends nothing at all.

load_queryables(t, collection_id, declared, dryrun=False, log=print)

Upsert a catalogue's declared queryables for its own collection. Returns the number of properties sent.

One document, one line, loaded in one transaction by pypgstac. The collection must already be registered: pgstac refuses a queryable naming a collection it lacks.

This is the bare load. pgstac replaces a global or shared row of each name with this collection's own, and nothing here checks what that took from other collections: run does, around this call.

describe_rules(target)

The asset rules a target applies, as a register run reports them.

Examples:

>>> target = Target(api="https://stac.example.invalid",
...                 collection_id="my-collection",
...                 bucket_url="https://bucket.example.invalid",
...                 require_asset="data", forbid_assets=["legacy"])
>>> describe_rules(target)
'require=data forbid=legacy'

run(target, mode, ids_file=None, dryrun=False, out_dir=None, log=print, warn=None)

Run one mode against one catalogue. Returns the exit status.

Every refusal is a RegisterError, raised before the first write wherever the check allows; it is printed and returns 1. dryrun still probes the API and the host (both read-only) and stops before the fetch (all, ids) or the audit (drift).