import sqlite3 from typing import Optional class ProviderBills: def __init__(self, db_path="instance/app_database.db"): self.db_path = db_path self._create_provider_bills_table() def _create_provider_bills_table(self): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS provider_bills ( id INTEGER PRIMARY KEY AUTOINCREMENT, number TEXT, date TEXT, provider_id INTEGER ); """) conn.commit() def add_provider_bills(self, number, date, provider_id): try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO provider_bills (number, date, provider_id) VALUES (?, ?, ?) """, (number, date, provider_id)) conn.commit() return True except sqlite3.IntegrityError: return False def get_all(self): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM provider_bills """,) rows = cursor.fetchall() result = [] if rows: for row in rows: buffer = { "id": row[0], "number": row[1], "date": row[2], "provider_id": row[3] } result.append(buffer) return result return [] def get_one(self, id): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM provider_bills WHERE id = ? """,(id, )) row = cursor.fetchone() result = [] if row: result = { "id": row[0], "number": row[1], "date": row[2], "provider_id": row[3] } return result return None def get_bill_by_provider_id(self, provider_id): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM provider_bills WHERE provider_id = ? """,(provider_id, )) row = cursor.fetchone() result = [] if row: result = { "id": row[0], "number": row[1], "date": row[2], "provider_id": row[3], "created_at": row[4] } return result return None def remove(self, id): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' DELETE FROM provider_bills WHERE id=?; ''', (id,)) conn.commit() def update(self, id, number, date, provider_id): try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" UPDATE provider_bills SET number = ?, date = ?, provider_id = ? WHERE id = ? """, (number, date, provider_id, id)) conn.commit() return True except Exception: return False