corrlib/corrlib/integrity.py
Justus Kuhlmann 23b5d066f7
All checks were successful
Mypy / mypy (push) Successful in 1m15s
Pytest / pytest (3.12) (push) Successful in 1m19s
Pytest / pytest (3.13) (push) Successful in 1m12s
Pytest / pytest (3.14) (push) Successful in 1m17s
Ruff / ruff (push) Successful in 1m0s
Mypy / mypy (pull_request) Successful in 1m13s
Pytest / pytest (3.12) (pull_request) Successful in 1m19s
Pytest / pytest (3.13) (pull_request) Successful in 1m11s
Pytest / pytest (3.14) (pull_request) Successful in 1m14s
Ruff / ruff (pull_request) Successful in 1m4s
make integrity checks accassible from cli
2026-04-17 16:34:30 +02:00

47 lines
1.4 KiB
Python

import datetime as dt
from pathlib import Path
from .tools import get_db_file
import pandas as pd
import sqlite3
def has_valid_times(result: pd.Series) -> bool:
# we expect created_at <= updated_at <= now
created_at = dt.datetime.fromisoformat(result['created_at'])
updated_at = dt.datetime.fromisoformat(result['updated_at'])
if created_at > updated_at:
return False
if updated_at > dt.datetime.now():
return False
return True
def are_keys_unique(db: Path, table: str, col: str) -> bool:
conn = sqlite3.connect(db)
c = conn.cursor()
c.execute(f"SELECT COUNT( DISTINCT CAST(path AS nvarchar(4000))), COUNT({col}) FROM {table};")
results = c.fetchall()[0]
conn.close()
return bool(results[0] == results[1])
def check_db_integrity(path: Path) -> None:
db = get_db_file(path)
if not are_keys_unique(path / db, 'backlogs', 'path'):
raise Exception("The paths the backlog table of the database links are not unique.")
search_expr = "SELECT * FROM 'backlogs'"
conn = sqlite3.connect(path / db)
results = pd.read_sql(search_expr, conn)
for _, result in results.iterrows():
if not has_valid_times(result):
raise ValueError(f"Result with id {result[id]} has wrong time signatures.")
print("DB:\t")
def full_integrity_check(path: Path) -> None:
check_db_integrity(path)
print("Full:\t")