🚧 Add import
This commit is contained in:
Jonas Bengtsson
+206
-1
@@ -38,7 +38,8 @@ from datetime import datetime, timedelta, timezone
|
||||
from typing import Any
|
||||
|
||||
import psycopg
|
||||
from fastapi import Depends, FastAPI, HTTPException, Query, Request, status
|
||||
from fastapi import Body, Depends, FastAPI, HTTPException, Query, Request, status
|
||||
from fastapi.responses import JSONResponse
|
||||
from psycopg import sql
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
@@ -847,6 +848,80 @@ def _validate_attribute_body(
|
||||
return key.strip(), data["value"], effective_from, note
|
||||
|
||||
|
||||
def _validate_import_rows(data: object, source: str) -> list[dict[str, Any]]:
|
||||
"""Validate an import body: a JSON array of exported-message objects
|
||||
(the GET /messages/export shape) — a single object is accepted too.
|
||||
|
||||
Required per row: "imei" (15 digits), "device" (non-empty string),
|
||||
"payload" (a JSON object). "topic" defaults to /import/<imei> when
|
||||
absent; "received_at"/"payload_time" may be absent or null (received_at
|
||||
then defaults to now at insert time). The exported "id" is deliberately
|
||||
IGNORED: ids are database-local BIGSERIAL values and the import lets
|
||||
Postgres assign fresh ones (re-importing the same file twice, or
|
||||
importing into the original database, must not collide).
|
||||
Raises ValueError with a descriptive message otherwise."""
|
||||
if isinstance(data, dict):
|
||||
rows = [data]
|
||||
elif isinstance(data, list):
|
||||
if not data:
|
||||
raise ValueError(f"{source}: the message list is empty")
|
||||
rows = data
|
||||
else:
|
||||
raise ValueError(
|
||||
f"{source}: expected a JSON array of message objects "
|
||||
"(or a single message object)"
|
||||
)
|
||||
validated: list[dict[str, Any]] = []
|
||||
for i, row in enumerate(rows):
|
||||
where = f"{source}: row {i + 1}"
|
||||
if not isinstance(row, dict):
|
||||
raise ValueError(f"{where}: each message must be a JSON object")
|
||||
imei = row.get("imei")
|
||||
if not isinstance(imei, str) or normalize_imei(imei.strip()) is None:
|
||||
raise ValueError(f"{where}: 'imei' must be a 15-digit string")
|
||||
imei = imei.strip()
|
||||
device = row.get("device")
|
||||
if not isinstance(device, str) or not device.strip():
|
||||
raise ValueError(f"{where}: 'device' must be a non-empty string")
|
||||
payload = row.get("payload")
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError(f"{where}: 'payload' must be a JSON object")
|
||||
parsed_times: dict[str, datetime | None] = {}
|
||||
for field in ("received_at", "payload_time"):
|
||||
raw = row.get(field)
|
||||
if raw is None:
|
||||
parsed_times[field] = None
|
||||
continue
|
||||
if not isinstance(raw, str):
|
||||
raise ValueError(f"{where}: '{field}' must be an ISO 8601 string or null")
|
||||
try:
|
||||
parsed_times[field] = _as_utc(datetime.fromisoformat(raw))
|
||||
except ValueError:
|
||||
raise ValueError(
|
||||
f"{where}: '{field}' is not a valid ISO 8601 timestamp"
|
||||
) from None
|
||||
topic = row.get("topic")
|
||||
if topic is None:
|
||||
topic = f"/import/{imei}"
|
||||
elif not isinstance(topic, str) or not topic:
|
||||
raise ValueError(f"{where}: 'topic' must be a non-empty string or absent")
|
||||
validated.append(
|
||||
{
|
||||
"imei": imei,
|
||||
"device": device.strip(),
|
||||
"payload": payload,
|
||||
"topic": topic,
|
||||
"received_at": parsed_times["received_at"],
|
||||
"payload_time": parsed_times["payload_time"],
|
||||
}
|
||||
)
|
||||
if len(validated) > 10000:
|
||||
raise ValueError(
|
||||
f"{source}: too many messages ({len(validated)}); the limit is 10000 per import"
|
||||
)
|
||||
return validated
|
||||
|
||||
|
||||
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."""
|
||||
@@ -1731,6 +1806,136 @@ def create_app(
|
||||
],
|
||||
}
|
||||
|
||||
@app.get(
|
||||
"/messages/export",
|
||||
tags=["data"],
|
||||
dependencies=[Depends(authenticate)],
|
||||
response_class=JSONResponse,
|
||||
)
|
||||
def export_messages(
|
||||
ids: list[int] | None = Query(
|
||||
None,
|
||||
description=(
|
||||
"Message ids to export (repeatable, e.g. ?ids=1&ids=2). "
|
||||
"Ordering of the export follows the ids given."
|
||||
),
|
||||
),
|
||||
) -> JSONResponse:
|
||||
"""Export messages AS THEY ARE STORED: bare mqtt_messages rows
|
||||
(no read-time settings merge, no resolved attributes). This is the
|
||||
shape an import accepts back, so exports round-trip. Selected ids
|
||||
are returned in the order they are given; unknown ids are silently
|
||||
skipped (a selection from a list that changed under the user should
|
||||
still export the rows that remain)."""
|
||||
if not ids:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail="select at least one message to export (repeatable ?ids=)",
|
||||
)
|
||||
if len(ids) > config.max_limit:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"too many ids requested ({len(ids)}); the limit is {config.max_limit}",
|
||||
)
|
||||
try:
|
||||
conn = psycopg.connect(config.dsn)
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
# ORDER BY position keeps the caller's ordering;
|
||||
# DISTINCT ON (id) deduplicates repeated ids, keeping
|
||||
# the FIRST occurrence's position.
|
||||
cur.execute(
|
||||
"""
|
||||
WITH selected(id, position) AS (
|
||||
SELECT DISTINCT ON (id) id, position
|
||||
FROM unnest(%s::bigint[]) WITH ORDINALITY AS t(id, position)
|
||||
ORDER BY id, position
|
||||
)
|
||||
SELECT m.id, m.received_at, m.topic, m.imei, m.device,
|
||||
m.payload, m.payload_time
|
||||
FROM mqtt_messages m
|
||||
JOIN selected s ON s.id = m.id
|
||||
ORDER BY s.position
|
||||
""",
|
||||
(ids,),
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
except psycopg.Error as exc:
|
||||
raise _db_unavailable(exc) from exc
|
||||
return JSONResponse(
|
||||
content=[
|
||||
{
|
||||
"id": row[0],
|
||||
"received_at": row[1].isoformat(),
|
||||
"topic": row[2],
|
||||
"imei": row[3],
|
||||
"device": row[4],
|
||||
"payload": row[5],
|
||||
"payload_time": row[6].isoformat() if row[6] is not None else None,
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
)
|
||||
|
||||
@app.post(
|
||||
"/messages/import",
|
||||
tags=["data"],
|
||||
dependencies=[Depends(authenticate), Depends(require_level("superuser"))],
|
||||
)
|
||||
def import_messages(body: Any = Body(...)) -> dict[str, Any]:
|
||||
"""Import messages previously exported via GET /messages/export:
|
||||
a JSON array of the exported rows (a single object is accepted).
|
||||
The rows are stored as-is (raw payload, original timestamps, original
|
||||
topic) so an export/import round trip is byte-identical on the data
|
||||
columns. The exported "id" is IGNORED — the database assigns fresh
|
||||
ids from its own BIGSERIAL, so re-importing the same file (or
|
||||
importing into the source database) never collides.
|
||||
Returns {"status": "imported", "imported": n}."""
|
||||
try:
|
||||
rows = _validate_import_rows(body, "request body")
|
||||
conn = psycopg.connect(config.dsn)
|
||||
try:
|
||||
with conn.transaction():
|
||||
with conn.cursor() as cur:
|
||||
# executemany goes through psycopg's pipeline mode ->
|
||||
# one batched multi-VALUES round trip, not n.
|
||||
# received_at is NOT NULL with DEFAULT now(), but an
|
||||
# explicit NULL bypasses the default - COALESCE
|
||||
# applies it for rows without a timestamp.
|
||||
cur.executemany(
|
||||
"""
|
||||
INSERT INTO mqtt_messages
|
||||
(received_at, topic, imei, device, payload,
|
||||
payload_time)
|
||||
VALUES (COALESCE(%s, now()), %s, %s, %s, %s, %s)
|
||||
""",
|
||||
[
|
||||
(
|
||||
row["received_at"],
|
||||
row["topic"],
|
||||
row["imei"],
|
||||
row["device"],
|
||||
Jsonb(row["payload"]),
|
||||
row["payload_time"],
|
||||
)
|
||||
for row in rows
|
||||
],
|
||||
)
|
||||
imported = len(rows)
|
||||
finally:
|
||||
conn.close()
|
||||
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("imported %d message(s)", imported)
|
||||
return {"status": "imported", "imported": imported}
|
||||
|
||||
return app
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user