| import sqlite3 |
| import os |
| from contextlib import contextmanager |
|
|
| DB_PATH = os.path.join(os.path.dirname(__file__), "app.db") |
|
|
| @contextmanager |
| def get_db_connection(): |
| conn = sqlite3.connect(DB_PATH) |
| conn.row_factory = sqlite3.Row |
| try: |
| yield conn |
| finally: |
| conn.close() |
|
|
| def execute_query(query: str, params: tuple = ()) -> list: |
| """Executes a READ-ONLY query and returns rows.""" |
| with get_db_connection() as conn: |
| cursor = conn.cursor() |
| cursor.execute(query, params) |
| rows = cursor.fetchall() |
| return [dict(row) for row in rows] |
|
|
