|
| 1 | +"""Extrinsic provenance for rows that enter the pipeline from outside. |
| 2 | +
|
| 3 | +Inside the pipeline, provenance is structural: a Computed table's row cannot |
| 4 | +exist unless its declared upstream exists, so the foreign-key graph *is* the |
| 5 | +lineage and nothing has to be recorded for it to hold. |
| 6 | +
|
| 7 | +At the boundary the structure runs out. Rows arrive in Entry tables from a |
| 8 | +person, an instrument, or a feed, and the framework has no way to say where |
| 9 | +they came from. This module supplies the slot and fills it. |
| 10 | +
|
| 11 | +The attribute is **framework-owned: no author ever writes it.** ``insert`` |
| 12 | +takes no provenance argument, and its content comes from three places, none of |
| 13 | +them the call site: |
| 14 | +
|
| 15 | +* **configuration** -- ``config.provenance.source``, set per deployment, naming |
| 16 | + the external system this process draws from; |
| 17 | +* **ambient connection state** -- the connecting user, host and database, the |
| 18 | + insert time, and the code version; |
| 19 | +* **ambient execution state** -- the ingesting table and key, when the insert |
| 20 | + runs inside a ``make()``. |
| 21 | +
|
| 22 | +That ownership is the point. A field an operator can set is weaker evidence |
| 23 | +than one the system sets, and nothing is left for a pipeline to neglect. |
| 24 | +Anything an author wants to record deliberately belongs in the data model as a |
| 25 | +visible attribute, where queries can reach it. |
| 26 | +""" |
| 27 | + |
| 28 | +import contextlib |
| 29 | +import contextvars |
| 30 | +import datetime |
| 31 | +import json |
| 32 | +from typing import Any |
| 33 | + |
| 34 | +#: Name of the hidden attribute. Hidden attributes are excluded from |
| 35 | +#: ``heading.attributes``, so this never appears in a query heading. |
| 36 | +PROV_ATTRIBUTE = "_prov" |
| 37 | + |
| 38 | +# Set by autopopulate around a make() call so that rows written to Entry tables |
| 39 | +# from inside an ingesting make() record what wrote them, which is what makes a |
| 40 | +# fanned-out row traceable without a foreign key. |
| 41 | +_ingesting: contextvars.ContextVar = contextvars.ContextVar("dj_ingesting", default=None) |
| 42 | + |
| 43 | + |
| 44 | +def set_ingesting(table_name, key, version=None): |
| 45 | + """Record the ``make()`` now executing; returns a token for ``reset_ingesting``. |
| 46 | +
|
| 47 | + Parameters |
| 48 | + ---------- |
| 49 | + table_name : str |
| 50 | + Full table name of the ingesting table. |
| 51 | + key : dict |
| 52 | + The key ``make()`` was called with. |
| 53 | + version : str, optional |
| 54 | + Code version, as resolved for the job. |
| 55 | + """ |
| 56 | + value = { |
| 57 | + "table": table_name, |
| 58 | + "key": {k: _jsonable(v) for k, v in (key or {}).items()}, |
| 59 | + } |
| 60 | + if version: |
| 61 | + value["version"] = version |
| 62 | + return _ingesting.set(value) |
| 63 | + |
| 64 | + |
| 65 | +def reset_ingesting(token): |
| 66 | + """Restore the ingesting context saved by :func:`set_ingesting`.""" |
| 67 | + if token is not None: |
| 68 | + _ingesting.reset(token) |
| 69 | + |
| 70 | + |
| 71 | +@contextlib.contextmanager |
| 72 | +def ingesting(table_name, key, version=None): |
| 73 | + """Scope :func:`set_ingesting` to a block.""" |
| 74 | + token = set_ingesting(table_name, key, version) |
| 75 | + try: |
| 76 | + yield |
| 77 | + finally: |
| 78 | + reset_ingesting(token) |
| 79 | + |
| 80 | + |
| 81 | +def _jsonable(value): |
| 82 | + """Render a key value in a form ``json.dumps`` accepts.""" |
| 83 | + if isinstance(value, (str, int, float, bool)) or value is None: |
| 84 | + return value |
| 85 | + if isinstance(value, (datetime.datetime, datetime.date, datetime.time)): |
| 86 | + return value.isoformat() |
| 87 | + if isinstance(value, bytes): |
| 88 | + return value.hex() |
| 89 | + return str(value) |
| 90 | + |
| 91 | + |
| 92 | +def build_payload(connection, config=None): |
| 93 | + """Assemble the provenance record for rows inserted on this connection. |
| 94 | +
|
| 95 | + Returns ``None`` when there is nothing worth recording, so that a row is |
| 96 | + left with ``NULL`` rather than an empty object. |
| 97 | + """ |
| 98 | + if config is None: |
| 99 | + from .settings import config as _config |
| 100 | + |
| 101 | + config = _config |
| 102 | + |
| 103 | + payload: dict[str, Any] = {"time": datetime.datetime.now(datetime.timezone.utc).isoformat()} |
| 104 | + |
| 105 | + conn_info = getattr(connection, "conn_info", None) or {} |
| 106 | + agent = {key: conn_info[key] for key in ("user", "host", "database_name") if conn_info.get(key) is not None} |
| 107 | + if agent: |
| 108 | + payload["agent"] = agent |
| 109 | + |
| 110 | + try: |
| 111 | + from .jobs import _get_job_version |
| 112 | + |
| 113 | + version = _get_job_version(getattr(connection, "_config", None) or config) |
| 114 | + except Exception: # version capture must never break an insert |
| 115 | + version = "" |
| 116 | + if version: |
| 117 | + payload["version"] = version |
| 118 | + |
| 119 | + source = config.provenance.source |
| 120 | + if source: |
| 121 | + payload["source"] = source |
| 122 | + |
| 123 | + context = _ingesting.get() |
| 124 | + if context: |
| 125 | + payload["context"] = context |
| 126 | + |
| 127 | + # Time alone says nothing about origin; without any of the other three this |
| 128 | + # is noise rather than a record. |
| 129 | + return payload if len(payload) > 1 else None |
| 130 | + |
| 131 | + |
| 132 | +def serialize(payload): |
| 133 | + """Render a payload for the ``json`` column. |
| 134 | +
|
| 135 | + ``default=str`` because ``config.provenance.source`` is deployment-supplied |
| 136 | + and typed ``dict[str, Any]``: a ``date`` or a ``Path`` in it would otherwise |
| 137 | + raise from inside every insert into every Entry table, with an error naming |
| 138 | + neither provenance nor the setting that caused it. |
| 139 | + """ |
| 140 | + return json.dumps(payload, default=str) |
0 commit comments