proitlab
fdb5d0de15
- db.py: abstraksi connect/query/queryone/replace_snapshot; %s → ? utk sqlite; replace snapshot mysql = CREATE IF NOT EXISTS + DELETE + INSERT (MVCC) - config/.env.example: VEX_DB_BACKEND (default sqlite) + VEX_DB_HOST/PORT/USER/PASS/NAME - get_voters.persist → db.replace_snapshot; dashboard baca via db.query - LIKE pakai ESCAPE '!' (backslash memutus literal string MySQL, B2); db.query selalu kembalikan list (PyMySQL tuple vs sqlite list, B3) - docker-compose.yml: kontainer uji mariadb:11 (databisnisid-mariadb, 127.0.0.1:3306) - test_mariadb.py: suite integrasi opt-in (skip bila bukan mysql) - SPEC.md §V17/T17/B2/B3; AGENTS.md runbook
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
"""Abstraksi penyimpanan: sqlite (default) atau MariaDB/MySQL (PyMySQL).
|
|
|
|
Pilih backend via config `VEX_DB_BACKEND` (env `.env`). Kedua backend memakai
|
|
skema `voters` yang sama; query berbagi sintaks ditulis dengan placeholder
|
|
`%s` yang diterjemahkan ke `?` untuk sqlite. Dashboard read-only (V10).
|
|
"""
|
|
|
|
import sqlite3
|
|
|
|
from config import (DB_BACKEND, DB_HOST, DB_NAME, DB_PASS, DB_PORT, DB_PATH,
|
|
DB_USER)
|
|
|
|
# Skema voters — sama untuk kedua backend (I.db)
|
|
_COLS = (
|
|
'owner VARCHAR(13) PRIMARY KEY, '
|
|
'weight TEXT NOT NULL, '
|
|
'staked DOUBLE NOT NULL, '
|
|
'scanned_at TEXT NOT NULL, '
|
|
'last_vote TEXT'
|
|
)
|
|
_SCHEMA = f'CREATE TABLE voters ({_COLS})'
|
|
_SCHEMA_IF_NOT_EXISTS = f'CREATE TABLE IF NOT EXISTS voters ({_COLS})'
|
|
|
|
|
|
def _translate(sql):
|
|
"""sqlite pakai `?`; mysql pakai `%s`. Ubah `%s` → `?` untuk sqlite."""
|
|
return sql.replace('%s', '?') if DB_BACKEND == 'sqlite' else sql
|
|
|
|
|
|
def connect():
|
|
"""Buka koneksi backend. sqlite: WAL agar pembaca tak terblokir."""
|
|
if DB_BACKEND == 'sqlite':
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.execute('PRAGMA journal_mode=WAL')
|
|
return conn
|
|
if DB_BACKEND == 'mysql':
|
|
import pymysql
|
|
return pymysql.connect(
|
|
host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASS,
|
|
database=DB_NAME, charset='utf8mb4',
|
|
)
|
|
raise ValueError(f'VEX_DB_BACKEND tak dikenal: {DB_BACKEND!r}')
|
|
|
|
|
|
def query(sql, params=()):
|
|
"""Jalankan SELECT → daftar baris (tuple), koneksi dibuka-tutup tiap panggil."""
|
|
conn = connect()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(_translate(sql), params)
|
|
return list(cur.fetchall()) # sqlite→list, PyMySQL→tuple: seragamkan
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def queryone(sql, params=()):
|
|
"""Jalankan SELECT → baris pertama (tuple) atau None."""
|
|
conn = connect()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(_translate(sql), params)
|
|
return cur.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def replace_snapshot(rows, scanned_at):
|
|
"""V7: ganti snapshot voters tiap skan (⊥ history/append).
|
|
|
|
sqlite: DROP+CREATE+INSERT (WAL melindungi pembaca).
|
|
mysql: CREATE IF NOT EXISTS + DELETE + INSERT dalam satu transaksi
|
|
→ MVCC memberi pembaca snapshot konsisten (⊥ torn read saat ganti harian).
|
|
"""
|
|
tuples = [(r['owner'], r['weight'], r['staked'], scanned_at, r.get('last_vote'))
|
|
for r in rows]
|
|
conn = connect()
|
|
try:
|
|
cur = conn.cursor()
|
|
if DB_BACKEND == 'sqlite':
|
|
cur.execute('DROP TABLE IF EXISTS voters')
|
|
cur.execute(_SCHEMA)
|
|
cur.executemany(
|
|
'INSERT INTO voters (owner, weight, staked, scanned_at, last_vote) '
|
|
'VALUES (?, ?, ?, ?, ?)',
|
|
tuples,
|
|
)
|
|
else:
|
|
cur.execute(_SCHEMA_IF_NOT_EXISTS)
|
|
cur.execute('DELETE FROM voters')
|
|
cur.executemany(
|
|
'INSERT INTO voters (owner, weight, staked, scanned_at, last_vote) '
|
|
'VALUES (%s, %s, %s, %s, %s)',
|
|
tuples,
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|