# critwire Learn: /learn/labs/e1.py # Run it after the lab loader (/learn/data/postgis/learn_curriculum.sql), on the same database, in the environment # /learn/labs/requirements.txt pins (the setup page, /learn/setup/, gives the uv commands): # python e1.py # It connects with libpq's PG* variables (PGHOST, PGPORT, PGDATABASE, PGUSER, PGPASSWORD) and reads the lab data from # LEARN_BASE (default https://critwire.com). Each block runs once, in page order, after py-connect and the boot setup; # an untested block is commented out under its reason. # %% py-connect # q: py-connect # status: ran import psycopg def connect(): """A connection to the lab database: plain table names find learn_curriculum's tables, then PostGIS in public.""" return psycopg.connect(options="-c search_path=learn_curriculum,public") # %% e1-request # q: e1-request # status: ran # after: py-connect import json BACKBONE = "d7dddbf4-2cf0-4f39-9b2a-bb099caae36c" # the GBIF Backbone Taxonomy, which critwire's ids.gbif come from BASIS = ["HUMAN_OBSERVATION", "OBSERVATION", "MACHINE_OBSERVATION", "PRESERVED_SPECIMEN", "MATERIAL_SAMPLE", "MATERIAL_CITATION", "OCCURRENCE"] def build_request(taxon_keys, user): """The download request: open records with coordinates, no geospatial issue, present, no living or fossil specimens.""" return { "creator": user, "sendNotification": False, "format": "SIMPLE_CSV", "checklistKey": BACKBONE, "predicate": { "type": "and", "predicates": [ {"type": "in", "key": "TAXON_KEY", "values": [str(k) for k in taxon_keys], "checklistKey": BACKBONE}, {"type": "in", "key": "LICENSE", "values": ["CC0_1_0", "CC_BY_4_0"]}, {"type": "equals", "key": "HAS_COORDINATE", "value": "true"}, {"type": "equals", "key": "HAS_GEOSPATIAL_ISSUE", "value": "false"}, {"type": "equals", "key": "OCCURRENCE_STATUS", "value": "PRESENT"}, {"type": "in", "key": "BASIS_OF_RECORD", "values": BASIS}, ], }, } with connect() as conn: keys = [k for (k,) in conn.execute("SELECT gbif_id FROM species WHERE gbif_id IS NOT NULL ORDER BY gbif_id")] request = build_request(keys, "your-gbif-username") taxa = request["predicate"]["predicates"][0] print(f"taxon keys: {len(taxa['values'])}, first {taxa['values'][0]}, last {taxa['values'][-1]}") print(f"checklistKey: request {request['checklistKey']}, TAXON_KEY {taxa['checklistKey']}") for p in request["predicate"]["predicates"][1:]: print(p["key"], p["type"], p.get("values", p.get("value"))) print("bytes:", len(json.dumps(request))) # %% e1-fetch # q: e1-fetch # status: ran import hashlib import http.client import os import time import urllib.error import urllib.request WAITS = [1, 2, 4, 8] # seconds before attempts 2 to 5 RETRY_AFTER_CAP = 60 # the longest wait a server's Retry-After can ask for here def transient(error): """Worth another try: a rate limit (429), a server error (5xx), a dropped or cut connection. A 404 or 403 is not.""" if isinstance(error, urllib.error.HTTPError): return error.code == 429 or error.code >= 500 return isinstance(error, (urllib.error.URLError, ConnectionError, TimeoutError, http.client.HTTPException)) def wait_for(error, attempt): """Seconds to wait before the next attempt: the server's Retry-After, in seconds and up to the cap, else the backoff.""" value = error.headers.get("Retry-After", "") if isinstance(error, urllib.error.HTTPError) else "" return min(int(value), RETRY_AFTER_CAP) if value.isdigit() else WAITS[attempt] def describe(error): if isinstance(error, urllib.error.HTTPError): value = error.headers.get("Retry-After") return f"HTTP {error.code}" + (f", Retry-After {value}" if value else "") return str(error) def fetch_file(url, path, size=None, sha256=None, log=lambda attempt, what: None): """Download url to path, whole. Each attempt writes a fresh path.part from its first byte, hashing as it arrives; a transient failure deletes the .part and starts the file again. Only a complete file whose size and sha256 equal what the publisher says (when given) gets its real name. Returns (bytes, sha256 hex).""" part = path + ".part" for attempt in range(len(WAITS) + 1): try: digest, got = hashlib.sha256(), 0 with urllib.request.urlopen(url, timeout=60) as r, open(part, "wb") as f: declared = r.headers.get("Content-Length") for chunk in iter(lambda: r.read(1 << 20), b""): digest.update(chunk) got += len(chunk) f.write(chunk) if declared is not None and got != int(declared): # urllib returns a cut body without an error raise ConnectionError(f"body cut at {got} of {declared} bytes") except Exception as e: if os.path.exists(part): os.remove(part) if isinstance(e, urllib.error.HTTPError): e.close() if not transient(e) or attempt == len(WAITS): raise wait = wait_for(e, attempt) log(attempt + 1, f"{describe(e)}; again in {wait} s") time.sleep(wait) continue if (size is not None and got != size) or (sha256 is not None and digest.hexdigest() != sha256): os.remove(part) # complete but not the publisher's file: retrying would fetch it again raise ValueError(f"{path}: {got} bytes, sha256 {digest.hexdigest()[:12]}; the publisher says " f"{size}, {(sha256 or '?')[:12]}") os.replace(part, path) log(attempt + 1, f"{got} bytes") return got, digest.hexdigest() # %% e1-retry # q: e1-retry # status: ran # after: e1-fetch import http.server import threading BODY = bytes(range(256)) * 400 # 102,400 bytes of made-up data SCRIPT = ["cut", "429", "whole"] # what the server does with each request, in order class Flaky(http.server.BaseHTTPRequestHandler): def do_GET(self): step = SCRIPT.pop(0) if step == "429": self.send_response(429) self.send_header("Retry-After", "1") self.send_header("Content-Length", "0") self.end_headers() return self.send_response(200) self.send_header("Content-Length", str(len(BODY))) self.end_headers() self.wfile.write(BODY if step == "whole" else BODY[:40_000]) # "cut": the connection closes 40,000 bytes in def log_message(self, *args): pass # no request log on stderr server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Flaky) # port 0: the system picks a free one threading.Thread(target=server.serve_forever, daemon=True).start() try: url = f"http://127.0.0.1:{server.server_port}/flaky.bin" got, digest = fetch_file(url, "flaky.bin", len(BODY), hashlib.sha256(BODY).hexdigest(), log=lambda attempt, what: print(f"attempt {attempt}: {what}")) finally: server.shutdown() server.server_close() print(f"flaky.bin: {got} bytes, sha256 {digest[:12]}; .part left: {os.path.exists('flaky.bin.part')}") # %% e1-checksum # q: e1-checksum # status: ran # after: e1-fetch import json import os BASE = os.environ.get("LEARN_BASE", "https://critwire.com") fetch_file(f"{BASE}/learn/data/manifest.json", "manifest.json") with open("manifest.json") as f: manifest = json.load(f) for name in ["occurrences.parquet", "datasets.csv"]: expected = manifest["files"][name] got, digest = fetch_file(f"{BASE}/learn/data/{name}", name, expected["bytes"], expected["sha256"]) print(f"{name}: {got} bytes, sha256 {digest[:12]} = manifest, {expected['rows']} rows") # %% e1-decide # q: e1-decide # status: ran from datetime import datetime, timedelta SKEW = timedelta(minutes=5) # a download listed a little before the intent may still be its POST's: clocks differ def same(a, b): """The same download request: what GBIF runs, not who asked.""" return all(a.get(k) == b.get(k) for k in ("format", "checklistKey", "predicate")) def decide(intent, request, listing=None): """Which download this run uses, before any POST. intent: what the last run saved just before its POST ({"request", "started", "key"}), or None. listing: the user's downloads, once asked for. Returns ("post",), ("resume", key), ("list",) (list the downloads, then decide again), ("adopt", key) or ("refuse", why).""" if intent is None: return ("post",) if not same(intent["request"], request): return ("refuse", "the intent is for another request: finish its download or delete the intent first") if intent["key"] is not None: return ("resume", intent["key"]) if listing is None: return ("list",) since = datetime.fromisoformat(intent["started"]) - SKEW keys = [d["key"] for d in listing if datetime.fromisoformat(d["created"]) >= since and same(d["request"], intent["request"])] if len(keys) == 1: return ("adopt", keys[0]) if keys: return ("refuse", f"{len(keys)} downloads match the POST: a person picks one") return ("refuse", "the POST may have made a download GBIF doesn't list yet: run again later") # Six made-up situations; "devil" stands in for e1-request's request, "other" for a changed one. devil = {"format": "SIMPLE_CSV", "checklistKey": "d7dddbf4", "predicate": {"type": "equals", "key": "TAXON_KEY", "value": "2435451"}} other = {**devil, "predicate": {"type": "equals", "key": "TAXON_KEY", "value": "2440897"}} unsaved = {"request": devil, "started": "2026-10-23T09:00:00+00:00", "key": None} # POSTed; the key never saved cases = { "first run": (None, None), "key saved": ({**unsaved, "key": "0041824-260928105237408"}, None), "key lost, GBIF lists one": (unsaved, [ {"key": "0041819-260928105237408", "created": "2026-10-23T08:40:00+00:00", "request": devil}, # older: not ours {"key": "0041824-260928105237408", "created": "2026-10-23T09:00:02+00:00", "request": devil}, {"key": "0041830-260928105237408", "created": "2026-10-23T09:03:00+00:00", "request": other}, # another request ]), "key lost, GBIF lists none": (unsaved, []), "key lost, GBIF lists two": (unsaved, [ {"key": "0041824-260928105237408", "created": "2026-10-23T09:00:02+00:00", "request": devil}, {"key": "0041825-260928105237408", "created": "2026-10-23T09:00:05+00:00", "request": devil}, ]), "request changed": (unsaved, None), } for name, (intent, listing) in cases.items(): request = other if name == "request changed" else devil step = decide(intent, request) steps = [step] if step == ("list",): steps.append(decide(intent, request, listing)) print(f"{name}: " + ", then ".join(": ".join(s) for s in steps)) # %% e1-submit # not run from this file: makes a real GBIF download with your account; the course's sample was made with the same steps. Run it by hand, as its page says. # q: e1-submit # status: untested: makes a real GBIF download with your account; the course's sample was made with the same steps # after: py-connect, e1-request, e1-fetch, e1-decide # import base64 # from datetime import timezone # # API = "https://api.gbif.org/v1" # USER, PASSWORD = os.environ["GBIF_USERNAME"], os.environ["GBIF_PASSWORD"] # from the environment, never in the code # AUTH = "Basic " + base64.b64encode(f"{USER}:{PASSWORD}".encode()).decode() # INTENT = "download-intent.json" # what this run asked for, saved before its POST # DEADLINE_S = 6 * 3600 # stop asking after 6 hours; a run started later resumes the same download # _last = 0.0 # # # def call(method, url, body=None, auth=False): # """One request, at most 2 a second. A GET retries what is transient, as fetch_file does; a POST never retries, # because a second POST is a second download.""" # global _last # headers = {"Accept": "application/json, text/plain"} # data = None # if body is not None: # data = json.dumps(body).encode() # headers["Content-Type"] = "application/json" # if auth: # headers["Authorization"] = AUTH # retries = len(WAITS) if method == "GET" else 0 # for attempt in range(retries + 1): # time.sleep(max(0.0, _last + 0.5 - time.monotonic())) # _last = time.monotonic() # try: # with urllib.request.urlopen(urllib.request.Request(url, data, headers, method=method), timeout=60) as r: # return r.read() # except Exception as e: # if not transient(e) or attempt == retries: # raise # time.sleep(wait_for(e, attempt)) # # # def save(intent): # """Write the intent whole: a .part first, then the rename, so a killed run never leaves half a file.""" # with open(INTENT + ".part", "w") as f: # json.dump(intent, f) # os.replace(INTENT + ".part", INTENT) # # # request = build_request(keys, USER) # intent = None # if os.path.exists(INTENT): # with open(INTENT) as f: # intent = json.load(f) # step = decide(intent, request) # if step == ("list",): # a run POSTed and stopped before saving the key: find its download, never POST again # listed = json.loads(call("GET", f"{API}/occurrence/download/user/{USER}?limit=50", auth=True))["results"] # step = decide(intent, request, [{"key": d["key"], "created": d["created"], "request": d["request"]} for d in listed]) # if step[0] == "refuse": # raise SystemExit(f"{INTENT}: {step[1]}") # if step == ("post",): # intent = {"request": request, "started": datetime.now(timezone.utc).isoformat(), "key": None} # save(intent) # before the POST: a run killed during it leaves the trace the next run reconciles # try: # key = call("POST", f"{API}/occurrence/download/request", request, auth=True).decode() # except urllib.error.HTTPError as e: # if e.code < 500: # GBIF answered no (a bad request, a wrong password, too many downloads): nothing was made # os.remove(INTENT) # raise # a 5xx or a dropped connection may have made one: the intent stays, and the next run lists # save({**intent, "key": key}) # else: # key = step[1] # if step[0] == "adopt": # saved before polling: a run stopped while waiting resumes this key, never lists again # save({**intent, "key": key}) # print("download key:", key) # # deadline = time.monotonic() + DEADLINE_S # while True: # a job, not a response: ask at most once a minute, and not forever # meta = json.loads(call("GET", f"{API}/occurrence/download/{key}")) # if meta["status"] == "SUCCEEDED": # break # if meta["status"] in ("KILLED", "FAILED", "CANCELLED"): # raise SystemExit(f"download {key} ended {meta['status']}; delete {INTENT} to ask again") # if time.monotonic() > deadline: # raise SystemExit(f"download {key} is still {meta['status']} after {DEADLINE_S // 3600} h; " # f"run again later: {INTENT} resumes it") # time.sleep(60) # stored = meta["request"] # if stored["checklistKey"] != BACKBONE or stored["predicate"]["predicates"][0]["checklistKey"] != BACKBONE: # raise SystemExit("GBIF didn't keep the pinned taxonomy") # the echo check: what ran is what was asked # print("DOI:", meta["doi"], "records:", meta["totalRecords"], "bytes:", meta["size"]) # # got, digest = fetch_file(meta["downloadLink"], f"{key}.zip", size=meta["size"]) # print("zip sha256:", digest) # keep it with the DOI: it is how a later run knows it has the same file # %% e1-ex # q: e1-ex # status: ran # after: e1-fetch import email.message from datetime import datetime, timezone from email.utils import parsedate_to_datetime def retry_after(value, now): """Seconds a Retry-After asks for, as a number of seconds or as an HTTP date, kept between 0 and the cap; None when it is neither.""" if value.isdigit(): seconds = int(value) else: try: when = parsedate_to_datetime(value) except (TypeError, ValueError): return None seconds = (when - now).total_seconds() return int(min(max(seconds, 0), RETRY_AFTER_CAP)) def wait_for(error, attempt, now=None): """e1-fetch's rule with both forms: the server's wait when it gives one that can be read, else the backoff.""" value = error.headers.get("Retry-After") if isinstance(error, urllib.error.HTTPError) else None seconds = retry_after(value, now or datetime.now(timezone.utc)) if value else None return WAITS[attempt] if seconds is None else seconds NOW = datetime(2026, 10, 29, 12, 0, 0, tzinfo=timezone.utc) for code, value in [(429, "1"), (503, "Thu, 29 Oct 2026 12:00:30 GMT"), (503, "Thu, 29 Oct 2026 11:00:00 GMT"), (503, "3600"), (503, "soon"), (404, None)]: headers = email.message.Message() if value is not None: headers["Retry-After"] = value error = urllib.error.HTTPError("https://example.org/file", code, "", headers, None) shown = f"HTTP {code}" + (f", Retry-After {value!r}" if value else "") print(f"{shown}: " + (f"again in {wait_for(error, 0, NOW)} s" if transient(error) else "not retried"))