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.

README.md

@@ -97,6 +97,27 @@ sleeps one second and `logic_func` is not called at all. And an exception raised
 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()`
@@ -150,7 +171,7 @@ sed 's/{{vless .TagName}}/0.0.0.dev0/' pyproject.toml.tmpl > pyproject.toml
 
 ## Error handling
 
-Every method in this library catches its own exceptions, prints a line, and returns an empty
+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

src/i7lib/__init__.py

@@ -1,5 +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", "WriteBatch", "run", "timeutils"]
+__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()