-
-
Notifications
You must be signed in to change notification settings - Fork 36
/
Copy pathlogic.py
37 lines (26 loc) · 1.12 KB
/
logic.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from __future__ import annotations
import sqlite3
def get_metric_mapping(db: sqlite3.Connection) -> dict[str, int]:
"""Gets a mapping from metric_name to metric_id."""
results = db.execute('SELECT name, id FROM metric_names').fetchall()
return dict(results)
def get_metric_has_data(db: sqlite3.Connection) -> dict[int, bool]:
res = db.execute('SELECT id, has_data FROM metric_names').fetchall()
return {k: bool(v) for k, v in res}
def get_previous_sha(db: sqlite3.Connection) -> str | None:
"""Gets the latest inserted SHA."""
result = db.execute(
# Use ROWID as a free, auto-incrementing, primary key.
'SELECT sha FROM metric_data ORDER BY ROWID DESC LIMIT 1',
).fetchone()
return result[0] if result else None
def get_metric_values(db: sqlite3.Connection, sha: str) -> dict[int, int]:
"""Gets the metric values from a specific commit.
:param db: Database object
:param text sha: A sha representing a single commit
"""
results = db.execute(
'SELECT metric_id, running_value FROM metric_data WHERE sha = ?',
(sha,),
)
return dict(results)