🚧 Add imei time varying attributes
This commit is contained in:
Jonas Bengtsson
+304
-3
@@ -7,7 +7,10 @@ PUT /configs/imei_map or — the way a management frontend should work —
|
||||
per device via the /devices/{imei} resource endpoints (single-row upsert,
|
||||
so concurrent edits of different devices never clobber each other).
|
||||
GET /messages merges each device's CURRENT settings document into the
|
||||
payload at read time; stored payload keys win on conflicts.
|
||||
payload at read time; stored payload keys win on conflicts. It also
|
||||
resolves each device's TIME-VARYING attributes (offsets etc., from the
|
||||
imei_attributes timeline) as of the message's device time and returns
|
||||
them as a sibling "attributes" field.
|
||||
|
||||
Authentication: app tokens stored as SHA-256 digests in the app_tokens
|
||||
table, passed either as the `X-App-Token` header or as an `app_token` query
|
||||
@@ -280,6 +283,118 @@ class ImeiMapStore:
|
||||
conn.close()
|
||||
|
||||
|
||||
class ImeiAttributesStore:
|
||||
"""Time-varying device attributes (e.g. calibration offsets) backed by
|
||||
the imei_attributes table: an append-only timeline where a new value
|
||||
is a new row. The value valid at time T is the row with the greatest
|
||||
effective_from <= T.
|
||||
|
||||
Attributes are read-time corrections only: the listener never reads
|
||||
this table, and rows may exist for IMEIs without an imei_map entry
|
||||
(there is no FK, like on mqtt_messages)."""
|
||||
|
||||
def __init__(self, dsn: str) -> None:
|
||||
self._dsn = dsn
|
||||
|
||||
def history(self, imei: str) -> dict[str, list[dict[str, Any]]]:
|
||||
"""Full timeline as {key: [versions]}, oldest first."""
|
||||
conn = psycopg.connect(self._dsn)
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT key, value, effective_from, note FROM imei_attributes "
|
||||
"WHERE imei = %s ORDER BY key, effective_from",
|
||||
(imei,),
|
||||
)
|
||||
result: dict[str, list[dict[str, Any]]] = {}
|
||||
for key, value, effective_from, note in cur.fetchall():
|
||||
result.setdefault(key, []).append(
|
||||
{"value": value, "effective_from": effective_from, "note": note}
|
||||
)
|
||||
return result
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def resolved(self, imei: str, at: datetime | None = None) -> dict[str, Any]:
|
||||
"""{key: value} as of `at` (None = now): for each key the version
|
||||
with the greatest effective_from <= at. Keys whose first version
|
||||
lies after `at` are absent (no value was in effect yet)."""
|
||||
conn = psycopg.connect(self._dsn)
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT DISTINCT ON (key) key, value
|
||||
FROM imei_attributes
|
||||
WHERE imei = %s AND effective_from <= COALESCE(%s, now())
|
||||
ORDER BY key, effective_from DESC
|
||||
""",
|
||||
(imei, at),
|
||||
)
|
||||
return dict(cur.fetchall())
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def append(
|
||||
self,
|
||||
imei: str,
|
||||
key: str,
|
||||
value: Any,
|
||||
effective_from: datetime | None = None,
|
||||
note: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Append one version (effective_from defaults to now()).
|
||||
|
||||
Re-using an exact existing (imei, key, effective_from) replaces
|
||||
that version in place - "edit the entry I just saved" for backdated
|
||||
corrections. Returns the stored version."""
|
||||
# autocommit: a single-statement write commits immediately (a plain
|
||||
# connection would roll back on close).
|
||||
conn = psycopg.connect(self._dsn, autocommit=True)
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO imei_attributes (imei, key, value, effective_from, note)
|
||||
VALUES (%s, %s, %s, COALESCE(%s, now()), %s)
|
||||
ON CONFLICT (imei, key, effective_from) DO UPDATE
|
||||
SET value = EXCLUDED.value, note = EXCLUDED.note, created_at = now()
|
||||
RETURNING key, value, effective_from, note
|
||||
""",
|
||||
(imei, key, Jsonb(value), effective_from, note),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
return {
|
||||
"key": row[0],
|
||||
"value": row[1],
|
||||
"effective_from": row[2],
|
||||
"note": row[3],
|
||||
}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def delete_latest(self, imei: str, key: str) -> bool:
|
||||
"""Delete ONLY the newest version of a key (typo/undo); the earlier
|
||||
history stays immutable. Returns False when the key is unknown."""
|
||||
conn = psycopg.connect(self._dsn, autocommit=True)
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
DELETE FROM imei_attributes
|
||||
WHERE imei = %s AND key = %s
|
||||
AND effective_from = (
|
||||
SELECT max(effective_from) FROM imei_attributes
|
||||
WHERE imei = %s AND key = %s
|
||||
)
|
||||
""",
|
||||
(imei, key, imei, key),
|
||||
)
|
||||
return cur.rowcount > 0
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _validate_token_body(data: dict[str, Any], source: str) -> dict[str, str]:
|
||||
"""Validate a {"tokens": {app: secret}} request body and return the
|
||||
inner tokens mapping. Raises ValueError with a descriptive message
|
||||
@@ -311,12 +426,53 @@ def _validate_single_token_body(data: dict[str, Any], source: str) -> str:
|
||||
return data["token"]
|
||||
|
||||
|
||||
def _validate_attribute_body(
|
||||
data: dict[str, Any], source: str
|
||||
) -> tuple[str, Any, datetime | None, str | None]:
|
||||
"""Validate a device-attribute append body and return
|
||||
(key, value, effective_from, note). effective_from None means "now".
|
||||
Raises ValueError with a descriptive message otherwise."""
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError(
|
||||
f'{source}: expected JSON on the form {{"key": ..., "value": ...}}'
|
||||
)
|
||||
key = data.get("key")
|
||||
if not isinstance(key, str) or not key.strip():
|
||||
raise ValueError(f'{source}: "key" must be a non-empty string')
|
||||
if "value" not in data or data["value"] is None:
|
||||
raise ValueError(f'{source}: "value" must carry a JSON value')
|
||||
effective_from: datetime | None = None
|
||||
if data.get("effective_from") is not None:
|
||||
raw = data["effective_from"]
|
||||
if not isinstance(raw, str):
|
||||
raise ValueError(f'{source}: "effective_from" must be an ISO 8601 string')
|
||||
try:
|
||||
effective_from = _as_utc(datetime.fromisoformat(raw))
|
||||
except ValueError:
|
||||
raise ValueError(
|
||||
f'{source}: "effective_from" is not a valid ISO 8601 timestamp'
|
||||
) from None
|
||||
note = data.get("note")
|
||||
if note is not None and not isinstance(note, str):
|
||||
raise ValueError(f'{source}: "note" must be a string when present')
|
||||
return key.strip(), data["value"], effective_from, note
|
||||
|
||||
|
||||
def _device_repr(imei: str, doc: dict[str, Any]) -> dict[str, Any]:
|
||||
"""API representation of a device: its settings document plus the
|
||||
IMEI, so a GET response can be edited and PUT straight back."""
|
||||
return {"imei": imei, **doc}
|
||||
|
||||
|
||||
def _version_repr(version: dict[str, Any]) -> dict[str, Any]:
|
||||
"""API representation of one device-attribute timeline version."""
|
||||
return {
|
||||
"value": version["value"],
|
||||
"effective_from": version["effective_from"].isoformat(),
|
||||
"note": version["note"],
|
||||
}
|
||||
|
||||
|
||||
def _env(name: str, default: str | None = None) -> str:
|
||||
value = os.environ.get(name)
|
||||
if not value:
|
||||
@@ -334,6 +490,7 @@ def _as_utc(value: datetime) -> datetime:
|
||||
def create_app(
|
||||
token_store: TokenStore | None = None,
|
||||
imei_store: ImeiMapStore | None = None,
|
||||
attributes_store: ImeiAttributesStore | None = None,
|
||||
) -> FastAPI:
|
||||
"""Build the API application.
|
||||
|
||||
@@ -358,6 +515,8 @@ def create_app(
|
||||
)
|
||||
if imei_store is None:
|
||||
imei_store = ImeiMapStore(dsn)
|
||||
if attributes_store is None:
|
||||
attributes_store = ImeiAttributesStore(dsn)
|
||||
|
||||
def authenticate(request: Request) -> None:
|
||||
token = request.headers.get("X-App-Token")
|
||||
@@ -510,6 +669,128 @@ def create_app(
|
||||
logger.info("device %s deleted", normalized)
|
||||
return {"status": "deleted", "imei": normalized}
|
||||
|
||||
# --- Time-varying device attributes (e.g. calibration offsets): an
|
||||
# append-only timeline per IMEI. The value valid at T is the newest
|
||||
# version with effective_from <= T; GET /messages resolves each
|
||||
# message's attributes against the device time (payload_time, with
|
||||
# received_at as the fallback). Attributes are API-only - the listener
|
||||
# never reads them, and they may exist for IMEIs without a settings
|
||||
# row (no FK).
|
||||
|
||||
@app.get(
|
||||
"/devices/{imei}/attributes/history",
|
||||
tags=["devices"],
|
||||
dependencies=[Depends(authenticate)],
|
||||
)
|
||||
def get_device_attributes_history(imei: str) -> dict[str, Any]:
|
||||
normalized = normalize_imei(imei)
|
||||
if normalized is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"{imei!r} is not a 15-digit IMEI",
|
||||
)
|
||||
try:
|
||||
timeline = attributes_store.history(normalized)
|
||||
except psycopg.Error as exc:
|
||||
raise _db_unavailable(exc) from exc
|
||||
return {
|
||||
"imei": normalized,
|
||||
"attributes": {
|
||||
key: [_version_repr(version) for version in versions]
|
||||
for key, versions in timeline.items()
|
||||
},
|
||||
}
|
||||
|
||||
@app.get("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)])
|
||||
def get_device_attributes(
|
||||
imei: str,
|
||||
at: datetime | None = Query(
|
||||
None,
|
||||
description=(
|
||||
"Resolve the values as of this time (ISO 8601, inclusive; naive "
|
||||
"values are treated as UTC). Default: now."
|
||||
),
|
||||
),
|
||||
) -> dict[str, Any]:
|
||||
normalized = normalize_imei(imei)
|
||||
if normalized is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"{imei!r} is not a 15-digit IMEI",
|
||||
)
|
||||
resolved_at = _as_utc(at) if at is not None else None
|
||||
try:
|
||||
values = attributes_store.resolved(normalized, resolved_at)
|
||||
except psycopg.Error as exc:
|
||||
raise _db_unavailable(exc) from exc
|
||||
result: dict[str, Any] = {"imei": normalized, "attributes": values}
|
||||
if resolved_at is not None:
|
||||
result["at"] = resolved_at.isoformat()
|
||||
return result
|
||||
|
||||
@app.put("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)])
|
||||
@app.post("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)])
|
||||
def add_device_attribute(imei: str, body: dict[str, Any]) -> dict[str, Any]:
|
||||
normalized = normalize_imei(imei)
|
||||
if normalized is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"{imei!r} is not a 15-digit IMEI",
|
||||
)
|
||||
try:
|
||||
key, value, effective_from, note = _validate_attribute_body(body, "request body")
|
||||
stored = attributes_store.append(normalized, key, value, effective_from, note)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=str(exc),
|
||||
) from exc
|
||||
except psycopg.Error as exc:
|
||||
raise _db_unavailable(exc) from exc
|
||||
logger.info(
|
||||
"attribute %r appended for %s (effective %s)",
|
||||
stored["key"],
|
||||
normalized,
|
||||
stored["effective_from"].isoformat(),
|
||||
)
|
||||
return {
|
||||
"imei": normalized,
|
||||
"key": stored["key"],
|
||||
"value": stored["value"],
|
||||
"effective_from": stored["effective_from"].isoformat(),
|
||||
"note": stored["note"],
|
||||
}
|
||||
|
||||
@app.delete(
|
||||
"/devices/{imei}/attributes/{key}",
|
||||
tags=["devices"],
|
||||
dependencies=[Depends(authenticate)],
|
||||
)
|
||||
def delete_device_attribute(imei: str, key: str) -> dict[str, Any]:
|
||||
normalized = normalize_imei(imei)
|
||||
if normalized is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"{imei!r} is not a 15-digit IMEI",
|
||||
)
|
||||
name = key.strip()
|
||||
if not name:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail="attribute key must be a non-empty string",
|
||||
)
|
||||
try:
|
||||
deleted = attributes_store.delete_latest(normalized, name)
|
||||
except psycopg.Error as exc:
|
||||
raise _db_unavailable(exc) from exc
|
||||
if not deleted:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"no versions of attribute {name!r} for device {normalized}",
|
||||
)
|
||||
logger.info("attribute %r latest version deleted for %s", name, normalized)
|
||||
return {"status": "deleted", "imei": normalized, "key": name}
|
||||
|
||||
# --- Token resources: add/rotate/remove ONE app token without
|
||||
# touching the other apps' rows.
|
||||
|
||||
@@ -676,10 +957,29 @@ def create_app(
|
||||
# concatenation makes the RIGHT operand win on duplicate keys: the
|
||||
# stored payload (the device's own data plus the ingest-time
|
||||
# imei/device snapshot) takes precedence over the settings doc.
|
||||
#
|
||||
# The LATERAL join resolves the device's TIME-VARYING attributes
|
||||
# (offsets etc.) as of the message's DEVICE time (payload_time,
|
||||
# received_at as the fallback): for each key the newest version
|
||||
# with effective_from <= that time. Returned as a sibling
|
||||
# "attributes" field so consumers apply the corrections themselves;
|
||||
# an absent key means no value was in effect then.
|
||||
query = sql.SQL(
|
||||
"SELECT m.id, m.received_at, m.topic, m.imei, m.device, "
|
||||
"COALESCE(im.meta, '{}'::jsonb) || m.payload, m.payload_time "
|
||||
"FROM mqtt_messages m LEFT JOIN imei_map im ON im.imei = m.imei"
|
||||
"COALESCE(im.meta, '{}'::jsonb) || m.payload, m.payload_time, "
|
||||
"attrs.resolved "
|
||||
"FROM mqtt_messages m "
|
||||
"LEFT JOIN imei_map im ON im.imei = m.imei "
|
||||
"LEFT JOIN LATERAL ("
|
||||
" SELECT COALESCE(jsonb_object_agg(latest.key, latest.value), '{}'::jsonb) AS resolved "
|
||||
" FROM ("
|
||||
" SELECT DISTINCT ON (a.key) a.key, a.value "
|
||||
" FROM imei_attributes a "
|
||||
" WHERE a.imei = m.imei "
|
||||
" AND a.effective_from <= COALESCE(m.payload_time, m.received_at) "
|
||||
" ORDER BY a.key, a.effective_from DESC "
|
||||
" ) latest"
|
||||
") attrs ON true"
|
||||
) + where + order_by
|
||||
|
||||
try:
|
||||
@@ -709,6 +1009,7 @@ def create_app(
|
||||
"device": row[4],
|
||||
"payload": row[5],
|
||||
"payload_time": row[6].isoformat() if row[6] is not None else None,
|
||||
"attributes": row[7],
|
||||
}
|
||||
for row in rows
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user