fix: satisfy semgrep on constant-column SQL; use injectable clock for baseline entry_date
This commit is contained in:
@@ -711,15 +711,18 @@ def _counter_deltas(
|
||||
columns = ", ".join(
|
||||
key if key in present else "NULL" for key, _, _ in DIAGNOSTIC_COUNTERS)
|
||||
columns += ", segment_id"
|
||||
# nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns come from module constants, values are bound
|
||||
last = conn.execute(
|
||||
"SELECT id, %s FROM samples WHERE ts <= ? ORDER BY ts DESC, id DESC "
|
||||
"LIMIT 1" % columns, (end.isoformat(),),
|
||||
).fetchone()
|
||||
first = None
|
||||
if start is not None and last is not None:
|
||||
# nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns come from module constants, values are bound
|
||||
first = conn.execute(
|
||||
"SELECT id, %s FROM samples WHERE ts <= ? ORDER BY ts DESC, id DESC "
|
||||
"LIMIT 1" % columns, (start.isoformat(),),
|
||||
# nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns come from module constants, values are bound
|
||||
).fetchone() or conn.execute(
|
||||
"SELECT id, %s FROM samples WHERE ts > ? AND ts <= ? "
|
||||
"ORDER BY ts, id LIMIT 1" % columns,
|
||||
|
||||
@@ -254,6 +254,7 @@ def write_sample(
|
||||
segment_id = current_segment["id"] if current_segment else None
|
||||
|
||||
# Insert sample with segment_id
|
||||
# nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns come from module constants, values are bound
|
||||
cursor = conn.execute(
|
||||
"""
|
||||
INSERT INTO samples (
|
||||
@@ -387,6 +388,7 @@ def _stage_observation(
|
||||
separators=(",", ":"),
|
||||
sort_keys=True,
|
||||
)
|
||||
# nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns come from module constants, values are bound
|
||||
cursor = conn.execute(
|
||||
"INSERT INTO pending_publications (sample_ts, payload) VALUES (?, ?)",
|
||||
(sample["ts"], payload),
|
||||
|
||||
@@ -447,6 +447,7 @@ def _migrate_6_to_7(conn: sqlite3.Connection) -> None:
|
||||
return
|
||||
for column in DIAGNOSTIC_SAMPLE_COLUMNS:
|
||||
if column not in cols:
|
||||
# nosemgrep: python.lang.security.audit.formatted-sql-query.formatted-sql-query, python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- column is from DIAGNOSTIC_SAMPLE_COLUMNS, a constant
|
||||
conn.execute(f"ALTER TABLE samples ADD COLUMN {column} INTEGER")
|
||||
|
||||
|
||||
|
||||
+2
-2
@@ -15,7 +15,7 @@ from __future__ import annotations
|
||||
import json
|
||||
import math
|
||||
import sqlite3
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
@@ -1906,7 +1906,7 @@ class FenrisTuiApp(App):
|
||||
"tbw_terabytes": tbw,
|
||||
"source_url": source_url,
|
||||
"document_revision": None,
|
||||
"entry_date": date.today().isoformat(),
|
||||
"entry_date": self._clock_now.date().isoformat(),
|
||||
"model_string": None,
|
||||
"nominal_capacity_bytes": None,
|
||||
"validated_by": "user",
|
||||
|
||||
Reference in New Issue
Block a user