124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
import sqlite3
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
@dataclass
|
|
class PaymentsModel:
|
|
id: Optional[int] = None
|
|
user_id: Optional[int] = None
|
|
name: Optional[str] = None
|
|
amount: Optional[float] = None
|
|
type: Optional[str] = None
|
|
created_at: Optional[str] = None
|
|
|
|
class Payments:
|
|
def __init__(self, db_path="instance/app_database.db"):
|
|
self.db_path = db_path
|
|
self._create_payment_table()
|
|
|
|
def _create_payment_table(self):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS payments (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
name TEXT,
|
|
amount REAL,
|
|
type TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
def add_payment(self, payment: PaymentsModel):
|
|
"""Adds a new payment entry to the database."""
|
|
try:
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO subscriptions_and_payments (user_id, name, amount, type)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(payment.user_id, payment.name, payment.amount, payment.type),
|
|
)
|
|
conn.commit()
|
|
return cursor.lastrowid
|
|
except sqlite3.IntegrityError:
|
|
return None
|
|
|
|
def get_payment(self, payment_id: int) -> PaymentsModel | None:
|
|
"""Retrieves a single payment entry by its ID."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM subscriptions_and_payments WHERE id = ?", (payment_id,))
|
|
row = cursor.fetchone()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return PaymentsModel(
|
|
id=row[0],
|
|
user_id=row[1],
|
|
name=row[2],
|
|
amount=row[3],
|
|
type=row[4],
|
|
created_at=row[5]
|
|
)
|
|
|
|
def get_all_payments(self):
|
|
"""Retrieves all payment entries from the database."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM subscriptions_and_payments")
|
|
rows = cursor.fetchall()
|
|
|
|
return [
|
|
PaymentsModel(
|
|
id=row[0],
|
|
user_id=row[1],
|
|
name=row[2],
|
|
amount=row[3],
|
|
type=row[4],
|
|
created_at=row[5]
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
def update_payment(self, payment_id: int, name: Optional[str] = None, amount: Optional[float] = None, type: Optional[str] = None):
|
|
"""Updates an existing payment entry."""
|
|
fields = []
|
|
params = []
|
|
|
|
if name is not None:
|
|
fields.append("name = ?")
|
|
params.append(name)
|
|
if amount is not None:
|
|
fields.append("amount = ?")
|
|
params.append(amount)
|
|
if type is not None:
|
|
fields.append("type = ?")
|
|
params.append(type)
|
|
|
|
if not fields:
|
|
return False # No fields to update
|
|
|
|
params.append(payment_id)
|
|
query = f"UPDATE subscriptions_and_payments SET {', '.join(fields)} WHERE id = ?"
|
|
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, tuple(params))
|
|
conn.commit()
|
|
return cursor.rowcount > 0
|
|
|
|
def delete_payment(self, payment_id: int):
|
|
"""Deletes a payment entry by its ID."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM subscriptions_and_payments WHERE id = ?", (payment_id,))
|
|
conn.commit()
|
|
return cursor.rowcount > 0 |