36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
"""Execute bundled analytical SQL against SQLite read-only, write real CSVs."""
|
|
import argparse
|
|
import csv
|
|
from pathlib import Path
|
|
import sqlite3
|
|
|
|
QUERY_NAMES = ('top_categories', 'daily_dynamics', 'anomalies')
|
|
|
|
|
|
def run_queries(db, output):
|
|
output = Path(output)
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
uri = Path(db).resolve().as_uri() + '?mode=ro'
|
|
counts = {}
|
|
with sqlite3.connect(uri, uri=True) as connection:
|
|
# One read snapshot keeps reports mutually consistent if a writer runs.
|
|
connection.execute('BEGIN')
|
|
for name in QUERY_NAMES:
|
|
sql = (Path(__file__).parent / f'{name}.sql').read_text(encoding='utf-8')
|
|
cursor = connection.execute(sql)
|
|
rows = cursor.fetchall()
|
|
with (output / f'{name}.csv').open('w', encoding='utf-8', newline='') as stream:
|
|
writer = csv.writer(stream)
|
|
writer.writerow([column[0] for column in cursor.description])
|
|
writer.writerows(rows)
|
|
counts[name] = len(rows)
|
|
return counts
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--db', type=Path, required=True)
|
|
parser.add_argument('--output', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
print(run_queries(args.db, args.output))
|