112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
import sqlite3
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
@dataclass
|
|
class SubscriptionsModel:
|
|
id: Optional[int] = None
|
|
user_id: Optional[int] = None
|
|
name: Optional[str] = None # Numele abonamentului (ex: "Abonament Lunar", "Abonament Anual")
|
|
pay_and_subs_id: Optional[int] = None # ID-ul plății asociate din tabela subscriptions_and_payments
|
|
mounts: Optional[int] = None
|
|
created_at: Optional[str] = None
|
|
|
|
class Subscriptions:
|
|
def __init__(self, db_path="instance/app_database.db"):
|
|
self.db_path = db_path
|
|
self._create_subscription_table()
|
|
|
|
def _create_subscription_table(self):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS subscriptions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
name TEXT,
|
|
pay_and_subs_id INTEGER,
|
|
mounts INTEGER,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (user_id) REFERENCES users(id),
|
|
FOREIGN KEY (pay_and_subs_id) REFERENCES subscriptions_and_payments(id)
|
|
);
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
def add_subscription(self, subscription: SubscriptionsModel):
|
|
"""Adds a new subscription entry to the database."""
|
|
try:
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO subscriptions (user_id, name, pay_and_subs_id, mounts)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(subscription.user_id, subscription.name, subscription.pay_and_subs_id, subscription.mounts),
|
|
)
|
|
conn.commit()
|
|
return cursor.lastrowid
|
|
except sqlite3.Error:
|
|
return None
|
|
|
|
def get_subscription(self, subscription_id: int) -> Optional[SubscriptionsModel]:
|
|
"""Retrieves a single subscription entry by its ID."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM subscriptions WHERE id = ?", (subscription_id,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return SubscriptionsModel(*row)
|
|
return None
|
|
|
|
def get_all_subscriptions(self):
|
|
"""Retrieves all subscription entries from the database."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM subscriptions")
|
|
rows = cursor.fetchall()
|
|
return [SubscriptionsModel(*row) for row in rows]
|
|
|
|
def get_subscriptions_by_user_id(self, user_id: int):
|
|
"""Retrieves all subscriptions for a specific user."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM subscriptions WHERE user_id = ?", (user_id,))
|
|
rows = cursor.fetchall()
|
|
return [SubscriptionsModel(*row) for row in rows]
|
|
|
|
def update_subscription(self, subscription_id: int, name: Optional[str] = None, pay_and_subs_id: Optional[int] = None, mounts: Optional[int] = None):
|
|
"""Updates an existing subscription entry."""
|
|
fields = []
|
|
params = []
|
|
if name is not None:
|
|
fields.append("name = ?")
|
|
params.append(name)
|
|
if pay_and_subs_id is not None:
|
|
fields.append("pay_and_subs_id = ?")
|
|
params.append(pay_and_subs_id)
|
|
if mounts is not None:
|
|
fields.append("mounts = ?")
|
|
params.append(mounts)
|
|
|
|
if not fields:
|
|
return False
|
|
|
|
params.append(subscription_id)
|
|
query = f"UPDATE subscriptions 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_subscription(self, subscription_id: int):
|
|
"""Deletes a subscription entry by its ID."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM subscriptions WHERE id = ?", (subscription_id,))
|
|
conn.commit()
|
|
return cursor.rowcount > 0 |