i7lib: send_email, email_status and read_history helpers over the sidecar (prompt 683, plan item 861)

New module sidecar_calls: retry on 503 and unreachable sidecar with a stable email id, SidecarError instead of silent failure, history series indexed by id and name with null gaps. README section and unittest tests with a stub sidecar (not run in the session: no action runs Python). Not tagged.

.gitignore

@@ -0,0 +1,6 @@
+dist/
+build/
+*.egg-info/
+__pycache__/
+*.pyc
+pyproject.toml

README.md

@@ -1,3 +1,186 @@
 # i7lib
 
-i7 library for scripts
+Python library for scripts running in an i7 job container. It gives a script three things:
+plant signals (read and write) through the job's sidecar, the customer database through the
+db-controller, and millisecond time helpers.
+
+## Availability
+
+Inside a job container the library is already importable: it is installed in the job's image,
+built from the `python-base` repository, whose `local-requirements.txt` names the `i7lib`
+version. Nothing is mounted from the gateway, so a new `i7lib` version reaches a script only
+through a new image tag. Outside a job, install the wheel published to this org's gitea PyPI
+index. The only hard dependency is `requests`; `python-dotenv` is optional and loaded if
+present, so a local `.env` can supply the two environment variables below.
+
+## Endpoints and environment
+
+| Variable | Default | Injected value in a job |
+|---|---|---|
+| `SIDECAR_URL` | `http://localhost:80` | `http://sidecar:80` |
+| `DB_CONTROLLER_URL` | `http://127.0.0.1:13213` | `http://sidecar:8081` |
+
+The defaults are for running a script on your own machine. In a job both are set for you, so
+a script never hardcodes them.
+
+## Quick start
+
+```python
+from i7lib import run
+
+ID_OUTPUT = 49
+
+def logic(plant):
+    temperature = plant.state.get("oven-temperature", 0.0)
+    batch = plant.new_write_batch()
+    batch.add(ID_OUTPUT, temperature * 2.0, "oven-output")
+    plant.send(batch)
+
+if __name__ == "__main__":
+    run(logic, interval_ms=1000, offset_ms=200)
+```
+
+`from i7lib import ...` re-exports the public API; `from i7lib.client import run` and the
+other submodule paths work equally well.
+
+## Signals
+
+Not every signal is available to every job. `plant.state` holds only the signals the job's
+sidecar exposes to it, keyed by both id and name. Which those are is configured for the job
+in its `signals.yml`; the names used as keys are the ones written there.
+
+A signal that is not exposed is simply absent from `plant.state` — reading it does not raise.
+Guard with `in` or `.get()` rather than assuming a key is there.
+
+## i7lib.client
+
+### `Plant`
+
+Holds the last state read from the sidecar.
+
+| Attribute | Contents |
+|---|---|
+| `state` | value by signal id **and** by signal name, for every signal that has a name |
+| `instants` | timestamp of each value, keyed the same way |
+| `signals` | the raw list returned by the sidecar |
+
+A signal present but carrying no value reads as `nan`; its instant defaults to `0`.
+
+**`read_all()`** — reads every subscribed signal once. Returns `state`, or `{}` on failure.
+
+**`read_sync(interval_ms, offset_ms)`** — blocks until the sidecar's next sample aligned to
+`interval_ms`, offset by `offset_ms`, then returns `state`, or `{}` on failure. This is the
+call `run` uses; it is what keeps a script in step with the sample clock instead of drifting.
+
+**`new_write_batch()`** — returns an empty `WriteBatch`.
+
+**`send(batch)`** — posts the batch. Does nothing if the batch is empty.
+
+### `WriteBatch`
+
+**`add(signal_id, value, name="")`** — queues one write. `signal_id` is coerced to `int`,
+`value` to `float`, `name` to `str`.
+
+A batch is **all or nothing**. If the sidecar does not recognise one of the ids in it, the
+entire batch is rejected and none of the writes land. An id it does recognise but which is
+not writable is skipped without failing the batch. Neither outcome is visible to the caller —
+see Error handling below.
+
+### `run(logic_func, interval_ms=1000, offset_ms=200)`
+
+The main loop. Creates a `Plant`, and on every cycle calls `read_sync` and then
+`logic_func(plant)`. It never returns.
+
+Two behaviours to know. When a read fails or comes back empty the cycle is skipped, the loop
+sleeps one second and `logic_func` is not called at all. And an exception raised inside
+`logic_func` is caught and printed, then the loop sleeps one second and continues — a script
+does not die on a bad tick, so a persistent bug shows up as repeated log lines rather than a
+crash.
+
+## i7lib.sidecar_calls
+
+Email and history calls through the job's sidecar (`docs/script-calls.md` in the i7sidecar repository has the routes). They need a sidecar image that carries the gateway link. Unlike the rest of this library, these three **raise `SidecarError`** on failure instead of printing and returning an empty result, because a lost email or an empty report must not look like success. `SidecarError` has `status` (HTTP status, `0` when the sidecar could not be reached), `code` and `message`.
+
+All three retry while the sidecar answers 503 or cannot be reached, pausing 1 s, 2 s, 4 s and up to 10 s, until `retry_seconds` has passed. Every other failure raises at once.
+
+**`send_email(to, subject, body_text="", body_html="", cc=None, bcc=None, attachments=None, sender=None, reply_to="", display_name="", ttl_seconds=86400, email_id=None, retry_seconds=60)`** — returns `{"emailId", "state"}` with state `queued`, meaning the gateway holds the email durably, not that it was delivered. `to` is an address or a list. `attachments` is a list of `(filename, bytes)`, `(filename, bytes, content_type)` or `{"filename", "content", "content_type"}`. `sender` is omitted for the platform's own sender, or `{"mode": "smtp", "host", "port", "security", "username", "fromAddress", "passwordSealed"}` with the sealed password as bytes. One `email_id` is generated per call and reused across its retries, so a retry after a lost answer never sends the email twice; pass your own `email_id` to make a repeated call of the whole script safe too. Limits: 50 recipients, 10 MiB of body and attachments together, `ttl_seconds` up to 604800 (7 days). Refusals: 400 `invalid`, 413 `too_large`, 429 `queue_full`.
+
+**`email_status(email_id, retry_seconds=30)`** — returns `{"state", "detail", "attempts", "updatedMs"}` with state `queued`, `sent`, `failed` or `expired`. An unknown id raises with status 404.
+
+**`read_history(signals, start_ms, end_ms, resolution="1min", retry_seconds=60)`** — `signals` is one or a list of signal names or ids, `resolution` is `1s`, `1min` or `1hr`. Returns a dict in which every series is found by signal id and by signal name; each series has `signalId`, `name`, `startMs`, `stepMs`, `values` (`None` for a gap) and `times` (the millisecond of each value). Only signals the job's `signals.yml` lists with `read: 1` can be read, otherwise 403. More than 200 signals raises 400, and an answer over 8 MiB raises 413.
+
+```python
+from i7lib import read_history, send_email
+from i7lib.timeutils import now
+
+history = read_history(["oven-temperature"], now() - 3600_000, now(), resolution="1min")
+series = history["oven-temperature"]
+send_email("plant@example.com", "Oven report", body_text=f"{len(series['values'])} samples")
+```
+
+## i7lib.dbutils
+
+### `Database()`
+
+Talks to the db-controller at `DB_CONTROLLER_URL`.
+
+| Method | Returns |
+|---|---|
+| `get_row(table, row_id)` | the row, or `None` |
+| `insert(table, data)` | `{'id': <new id>}`, or `None` |
+| `query(table, filters=None, order_by=None, limit=100)` | list of rows, or `[]` |
+| `update(table, row_id, data)` | `True` / `False` |
+| `list_tables()` | list of table names, or `[]` |
+| `list_columns(table)` | list of columns, or `[]` |
+
+Table and column names come from the customer schema, not from this library.
+
+These calls set no timeout, unlike the signal calls, so a db-controller that accepts a
+connection and then stalls will block the script. Keep database work off the hot path of a
+fast `interval_ms` loop.
+
+## i7lib.timeutils
+
+**`now()`** — current time in Unix milliseconds, the format used throughout the platform.
+
+**`now_sec()`** — the same, floored to the second.
+
+**`to_datetime(unix_milli)`** — converts back to a local `datetime`, for logging.
+
+## Versioning and the pyproject template
+
+There is no `pyproject.toml` in this repository — it is generated from
+`pyproject.toml.tmpl`, whose version line reads `version = "{{vless .TagName}}"`.
+gitea-sidecar renders every `*.tmpl` in the source to its name minus the suffix before
+building the wheel, so a tag of `v0.1.5` publishes version `0.1.5`; `vless` strips the
+leading `v` that git wants and PEP 440 forbids. The tag is the only place a version is
+written, and the published wheel cannot disagree with the tag that produced it.
+
+Do not put a hardcoded version back into the template: the build would keep producing that
+one version, and the next tag would fail with a 409 because the package file already
+exists.
+
+**Building locally** needs the same render step, since the file the build tools look for is
+not in the repository:
+
+```bash
+sed 's/{{vless .TagName}}/0.0.0.dev0/' pyproject.toml.tmpl > pyproject.toml
+```
+
+`pyproject.toml` is gitignored, so a local copy never reaches a build.
+
+## Error handling
+
+Every method in this library, except the three in `i7lib.sidecar_calls`, catches its own exceptions, prints a line, and returns an empty
+or falsy result. Nothing raises, and no method reports which signal or row failed.
+
+The practical consequences: `read_all` and `read_sync` returning `{}` mean "failed" and
+"nothing subscribed" alike; `query` and `list_tables` returning `[]` mean "failed" and "no
+rows" alike; `send` returning normally does **not** mean the writes were applied. A batch
+rejected in full for one unknown id looks exactly like a successful send from inside the
+script.
+
+So when a value is not moving, check the job's own output first — that is where this library
+prints its errors. If nothing appears there, the write reached the platform and was rejected
+or dropped further in, which is visible only in the platform-side logs. Do not infer success
+from the absence of an exception.

pyproject.toml.tmpl

@@ -0,0 +1,17 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "i7lib"
+version = "{{vless .TagName}}"
+description = "i7 library for scripts"
+readme = "README.md"
+requires-python = ">=3.8"
+dependencies = ["requests>=2.31"]
+
+[project.optional-dependencies]
+dotenv = ["python-dotenv>=1.0"]
+
+[tool.setuptools.packages.find]
+where = ["src"]

src/i7lib/__init__.py

@@ -0,0 +1,16 @@
+from . import timeutils
+from .client import Plant, WriteBatch, run
+from .dbutils import Database
+from .sidecar_calls import SidecarError, email_status, read_history, send_email
+
+__all__ = [
+    "Database",
+    "Plant",
+    "SidecarError",
+    "WriteBatch",
+    "email_status",
+    "read_history",
+    "run",
+    "send_email",
+    "timeutils",
+]

src/i7lib/sidecar_calls.py

@@ -0,0 +1,151 @@
+import base64
+import os
+import time
+import uuid
+
+import requests
+
+try:
+    from dotenv import load_dotenv
+    load_dotenv()
+except ImportError:
+    pass
+
+SIDECAR_URL = os.getenv("SIDECAR_URL", "http://localhost:80")
+
+RETRY_STATUSES = (503,)
+SEND_TIMEOUT = (5, 50)
+STATUS_TIMEOUT = (5, 25)
+HISTORY_TIMEOUT = (5, 80)
+FIRST_PAUSE = 1.0
+MAX_PAUSE = 10.0
+
+_sleep = time.sleep
+_now = time.monotonic
+
+
+class SidecarError(Exception):
+    def __init__(self, status, code, message):
+        super().__init__(f"{status} {code}: {message}")
+        self.status = status
+        self.code = code
+        self.message = message
+
+
+def _encode_bytes(value):
+    if isinstance(value, (bytes, bytearray)):
+        return base64.b64encode(bytes(value)).decode("ascii")
+    return value
+
+
+def _encode_attachment(item):
+    if isinstance(item, dict):
+        filename = item["filename"]
+        content = item["content"]
+        content_type = item.get("content_type", "")
+    else:
+        filename, content, *rest = item
+        content_type = rest[0] if rest else ""
+    return {
+        "filename": filename,
+        "contentType": content_type,
+        "dataBase64": _encode_bytes(content),
+    }
+
+
+def _encode_sender(sender):
+    if not sender:
+        return {"mode": "izazpi"}
+    encoded = dict(sender)
+    if "passwordSealed" in encoded:
+        encoded["passwordSealed"] = _encode_bytes(encoded["passwordSealed"])
+    return encoded
+
+
+def _failure_from(response):
+    try:
+        body = response.json()
+    except ValueError:
+        return SidecarError(response.status_code, "bad_response", "the sidecar answered something that is not JSON")
+    return SidecarError(response.status_code, body.get("errorCode") or "error", body.get("errorMessage") or "")
+
+
+def _pause_for(attempt):
+    return min(FIRST_PAUSE * (2 ** attempt), MAX_PAUSE)
+
+
+def _attempt(method, url, timeout, payload):
+    try:
+        response = requests.request(method, url, json=payload, timeout=timeout)
+    except requests.exceptions.RequestException as error:
+        return None, SidecarError(0, "sidecar_unreachable", str(error))
+    if response.status_code == 200:
+        return response.json().get("data"), None
+    return None, _failure_from(response)
+
+
+def _is_retryable(failure):
+    return failure.status in RETRY_STATUSES or failure.code == "sidecar_unreachable"
+
+
+def _call(method, path, timeout, retry_seconds, payload=None):
+    url = f"{SIDECAR_URL}{path}"
+    deadline = _now() + retry_seconds
+    attempt = 0
+    while True:
+        data, failure = _attempt(method, url, timeout, payload)
+        if failure is None:
+            return data
+        if not _is_retryable(failure) or _now() >= deadline:
+            raise failure
+        _sleep(_pause_for(attempt))
+        attempt += 1
+
+
+def send_email(to, subject, body_text="", body_html="", cc=None, bcc=None, attachments=None,
+               sender=None, reply_to="", display_name="", ttl_seconds=86400, email_id=None,
+               retry_seconds=60):
+    email_id = email_id or str(uuid.uuid4())
+    payload = {
+        "emailId": email_id,
+        "to": list(to) if not isinstance(to, str) else [to],
+        "cc": list(cc or []),
+        "bcc": list(bcc or []),
+        "subject": subject,
+        "bodyText": body_text,
+        "bodyHtml": body_html,
+        "attachments": [_encode_attachment(item) for item in (attachments or [])],
+        "sender": _encode_sender(sender),
+        "replyTo": reply_to,
+        "displayName": display_name,
+        "ttlSeconds": int(ttl_seconds),
+    }
+    return _call("POST", "/api/email/send/", SEND_TIMEOUT, retry_seconds, payload)
+
+
+def email_status(email_id, retry_seconds=30):
+    return _call("GET", f"/api/email/status/{email_id}/", STATUS_TIMEOUT, retry_seconds)
+
+
+def _history_path(signals, start_ms, end_ms, resolution):
+    names = ",".join(str(signal) for signal in signals)
+    return f"/api/history/?signals={requests.utils.quote(names, safe=',')}&start={int(start_ms)}&end={int(end_ms)}&resolution={resolution}"
+
+
+def _index_series(series):
+    series = dict(series)
+    series["times"] = [series["startMs"] + index * series["stepMs"] for index in range(len(series["values"]))]
+    return series
+
+
+def read_history(signals, start_ms, end_ms, resolution="1min", retry_seconds=60):
+    if isinstance(signals, (str, int)):
+        signals = [signals]
+    data = _call("GET", _history_path(signals, start_ms, end_ms, resolution), HISTORY_TIMEOUT, retry_seconds)
+    result = {}
+    for item in data["series"]:
+        indexed = _index_series(item)
+        result[indexed["signalId"]] = indexed
+        if indexed.get("name"):
+            result[indexed["name"]] = indexed
+    return result

tests/test_sidecar_calls.py

@@ -0,0 +1,147 @@
+import base64
+import json
+import sys
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from i7lib import sidecar_calls
+from i7lib.sidecar_calls import SidecarError, email_status, read_history, send_email
+
+
+class Script:
+    def __init__(self, answers):
+        self.answers = list(answers)
+        self.seen = []
+
+
+def make_handler(script):
+    class Handler(BaseHTTPRequestHandler):
+        def _answer(self):
+            length = int(self.headers.get("Content-Length") or 0)
+            body = self.rfile.read(length) if length else b""
+            script.seen.append((self.command, self.path, json.loads(body) if body else None))
+            status, payload = script.answers.pop(0) if len(script.answers) > 1 else script.answers[0]
+            raw = json.dumps(payload).encode()
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(raw)))
+            self.end_headers()
+            self.wfile.write(raw)
+
+        do_GET = _answer
+        do_POST = _answer
+
+        def log_message(self, *args):
+            pass
+
+    return Handler
+
+
+def ok(data):
+    return 200, {"data": data, "statusCode": 200, "errorCode": "", "errorMessage": ""}
+
+
+def failure(status, code, message="m"):
+    return status, {"data": None, "statusCode": status, "errorCode": code, "errorMessage": message}
+
+
+class SidecarCallsTest(unittest.TestCase):
+    def serve(self, answers):
+        script = Script(answers)
+        server = HTTPServer(("127.0.0.1", 0), make_handler(script))
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(server.shutdown)
+        self.addCleanup(server.server_close)
+        sidecar_calls.SIDECAR_URL = f"http://127.0.0.1:{server.server_address[1]}"
+        return script
+
+    def setUp(self):
+        self.pauses = []
+        sidecar_calls._sleep = self.pauses.append
+        self.addCleanup(lambda: setattr(sidecar_calls, "_sleep", __import__("time").sleep))
+
+    def test_send_email_posts_the_encoded_request_and_returns_the_answer(self):
+        script = self.serve([ok({"emailId": "x", "state": "queued"})])
+        answer = send_email(
+            "a@example.com", "Report", body_text="hi",
+            attachments=[("r.csv", b"a,b", "text/csv")],
+            sender={"mode": "smtp", "host": "h", "port": 587, "passwordSealed": b"\x01\x02"},
+        )
+        self.assertEqual(answer["state"], "queued")
+        method, path, body = script.seen[0]
+        self.assertEqual((method, path), ("POST", "/api/email/send/"))
+        self.assertEqual(body["to"], ["a@example.com"])
+        self.assertEqual(body["attachments"][0]["dataBase64"], base64.b64encode(b"a,b").decode())
+        self.assertEqual(body["sender"]["passwordSealed"], base64.b64encode(b"\x01\x02").decode())
+        self.assertTrue(body["emailId"])
+
+    def test_send_email_retries_on_503_with_the_same_email_id(self):
+        script = self.serve([failure(503, "gateway_unreachable"), failure(503, "gateway_unreachable"), ok({"emailId": "x", "state": "queued"})])
+        answer = send_email(["a@example.com"], "Report", retry_seconds=60)
+        self.assertEqual(answer["state"], "queued")
+        ids = {body["emailId"] for _, _, body in script.seen}
+        self.assertEqual(len(script.seen), 3)
+        self.assertEqual(len(ids), 1)
+        self.assertEqual(self.pauses, [1.0, 2.0])
+
+    def test_send_email_gives_up_when_the_retry_time_is_used(self):
+        self.serve([failure(503, "gateway_unreachable")])
+        with self.assertRaises(SidecarError) as caught:
+            send_email(["a@example.com"], "Report", retry_seconds=0)
+        self.assertEqual(caught.exception.status, 503)
+        self.assertEqual(caught.exception.code, "gateway_unreachable")
+
+    def test_send_email_does_not_retry_refusals_that_will_not_change(self):
+        script = self.serve([failure(400, "invalid", "bad address")])
+        with self.assertRaises(SidecarError) as caught:
+            send_email(["bad"], "Report")
+        self.assertEqual(caught.exception.code, "invalid")
+        self.assertEqual(len(script.seen), 1)
+        self.assertEqual(self.pauses, [])
+
+    def test_a_full_outbox_and_an_oversized_email_raise_at_once(self):
+        for status, code in ((429, "queue_full"), (413, "too_large")):
+            script = self.serve([failure(status, code)])
+            with self.assertRaises(SidecarError) as caught:
+                send_email(["a@example.com"], "Report")
+            self.assertEqual((caught.exception.status, caught.exception.code), (status, code))
+            self.assertEqual(len(script.seen), 1)
+
+    def test_email_status_returns_the_state_and_raises_not_found(self):
+        self.serve([ok({"state": "sent", "detail": "", "attempts": 1, "updatedMs": 5})])
+        self.assertEqual(email_status("e1")["state"], "sent")
+        self.serve([failure(404, "not_found")])
+        with self.assertRaises(SidecarError) as caught:
+            email_status("nope")
+        self.assertEqual(caught.exception.status, 404)
+
+    def test_read_history_indexes_series_by_name_and_id_with_times(self):
+        series = {"signalId": 1, "name": "cpu", "startMs": 60000, "stepMs": 60000, "values": [1.5, None, 3]}
+        script = self.serve([ok({"series": [series]})])
+        result = read_history(["cpu", 2], 0, 240000, resolution="1min")
+        self.assertIs(result["cpu"], result[1])
+        self.assertEqual(result["cpu"]["values"], [1.5, None, 3])
+        self.assertEqual(result["cpu"]["times"], [60000, 120000, 180000])
+        _, path, _ = script.seen[0]
+        self.assertEqual(path, "/api/history/?signals=cpu,2&start=0&end=240000&resolution=1min")
+
+    def test_read_history_raises_403_for_a_signal_not_entitled(self):
+        self.serve([failure(403, "signal_not_entitled")])
+        with self.assertRaises(SidecarError) as caught:
+            read_history("param", 0, 60000)
+        self.assertEqual(caught.exception.status, 403)
+
+    def test_a_sidecar_that_is_down_is_retried_then_raised(self):
+        sidecar_calls.SIDECAR_URL = "http://127.0.0.1:1"
+        with self.assertRaises(SidecarError) as caught:
+            email_status("e1", retry_seconds=0)
+        self.assertEqual(caught.exception.code, "sidecar_unreachable")
+
+
+if __name__ == "__main__":
+    unittest.main()