From 0d026e5d38b169fe6ba9d8208700879df7f04c34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?I=C3=B1igo=20Etxaniz?= Date: Wed, 28 Oct 2026 11:11:52 +0100 Subject: [PATCH] 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 | 23 +++++- src/i7lib/__init__.py | 13 +++- src/i7lib/sidecar_calls.py | 151 ++++++++++++++++++++++++++++++++++++ tests/test_sidecar_calls.py | 147 +++++++++++++++++++++++++++++++++++ 4 files changed, 332 insertions(+), 2 deletions(-) create mode 100644 src/i7lib/sidecar_calls.py create mode 100644 tests/test_sidecar_calls.py diff --git a/README.md b/README.md index 6f26c2f..3dec1a2 100644 --- a/README.md +++ b/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 diff --git a/src/i7lib/__init__.py b/src/i7lib/__init__.py index eca5cb3..41aa58b 100644 --- a/src/i7lib/__init__.py +++ b/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", +] diff --git a/src/i7lib/sidecar_calls.py b/src/i7lib/sidecar_calls.py new file mode 100644 index 0000000..e04cc98 --- /dev/null +++ b/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 diff --git a/tests/test_sidecar_calls.py b/tests/test_sidecar_calls.py new file mode 100644 index 0000000..09ccc59 --- /dev/null +++ b/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()