62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Small TDengine REST helpers for local smoke verifiers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
def sql_literal(value: str) -> str:
|
|
return "'" + (value or "").replace("'", "''") + "'"
|
|
|
|
|
|
def raw_frame_count_sql(
|
|
*,
|
|
protocol: str,
|
|
vehicle_key: str = "",
|
|
vin: str = "",
|
|
phone: str = "",
|
|
raw_uri: str = "",
|
|
) -> str:
|
|
if not protocol or not protocol.strip():
|
|
raise ValueError("protocol is required")
|
|
filters = ["protocol = " + sql_literal(protocol.strip().upper())]
|
|
optional_filters = {
|
|
"vehicle_key": vehicle_key,
|
|
"vin": vin,
|
|
"phone": phone,
|
|
"raw_uri": raw_uri,
|
|
}
|
|
present = False
|
|
for column, value in optional_filters.items():
|
|
if value and value.strip():
|
|
filters.append(column + " = " + sql_literal(value.strip()))
|
|
present = True
|
|
if not present:
|
|
raise ValueError("at least one identity or raw_uri filter is required")
|
|
return "SELECT COUNT(*) FROM raw_frames WHERE " + " AND ".join(filters)
|
|
|
|
|
|
def count_from_response(response: dict[str, Any]) -> int:
|
|
if response.get("code") != 0:
|
|
raise RuntimeError(f"tdengine query failed: {response}")
|
|
data = response.get("data") or []
|
|
if not data or not data[0]:
|
|
return 0
|
|
return int(data[0][0])
|
|
|
|
|
|
def query_count(rest_url: str, username: str, password: str, sql: str, timeout: float) -> int:
|
|
request = urllib.request.Request(rest_url, data=sql.encode("utf-8"), method="POST")
|
|
if username:
|
|
token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
|
|
request.add_header("Authorization", "Basic " + token)
|
|
request.add_header("Content-Type", "text/plain; charset=utf-8")
|
|
request.add_header("Accept", "application/json")
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
payload = json.loads(response.read().decode("utf-8"))
|
|
return count_from_response(payload)
|