163 lines
4.7 KiB
Python
163 lines
4.7 KiB
Python
# database/models/setting.py
|
|
import sqlite3
|
|
import time
|
|
from database.db import get_connection
|
|
|
|
class Setting:
|
|
|
|
# --- Constants for setting keys ---
|
|
AUTO_LISTING = "AUTO_LISTING"
|
|
MAX_CONCURRENT_LISTING = "MAX_CONCURRENT_LISTING"
|
|
LISTING_INTERVAL_SECONDS = "LISTING_INTERVAL_SECONDS"
|
|
MAX_CONCURRENT_LISTING = "MAX_CONCURRENT_LISTING"
|
|
|
|
|
|
@staticmethod
|
|
def all():
|
|
conn = get_connection()
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT * FROM settings")
|
|
rows = cur.fetchall()
|
|
conn.close()
|
|
return [dict(row) for row in rows]
|
|
|
|
@staticmethod
|
|
def get_by_id(setting_id):
|
|
conn = get_connection()
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT * FROM settings WHERE id = ?", (setting_id,))
|
|
row = cur.fetchone()
|
|
conn.close()
|
|
return dict(row) if row else None
|
|
|
|
@staticmethod
|
|
def get_by_key(key):
|
|
conn = get_connection()
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"SELECT * FROM settings WHERE key = ?",
|
|
(key,)
|
|
)
|
|
row = cur.fetchone()
|
|
conn.close()
|
|
if row:
|
|
return dict(row)
|
|
return None
|
|
|
|
@staticmethod
|
|
def create(key, value):
|
|
existing = Setting.get_by_key(key)
|
|
if existing:
|
|
return existing
|
|
|
|
created_at = updated_at = int(time.time())
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO settings (key, value, created_at, updated_at) VALUES (?, ?, ?, ?)",
|
|
(key, value, created_at, updated_at)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return Setting.get_by_key(key)
|
|
|
|
|
|
@staticmethod
|
|
def delete(setting_id):
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM settings WHERE id = ?", (setting_id,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
@staticmethod
|
|
def get_paginated(offset, limit, filters=None, sort_by="id", sort_order="ASC"):
|
|
filters = filters or {}
|
|
params = []
|
|
|
|
# Lấy luôn type
|
|
base_query = "SELECT id, key, value, type, created_at, updated_at FROM settings"
|
|
where_clauses = []
|
|
|
|
# --- Filters ---
|
|
if "key" in filters and filters["key"].strip():
|
|
where_clauses.append("key LIKE ?")
|
|
params.append(f"%{filters['key'].strip()}%")
|
|
|
|
if "value" in filters and filters["value"].strip():
|
|
where_clauses.append("value LIKE ?")
|
|
params.append(f"%{filters['value'].strip()}%")
|
|
|
|
if where_clauses:
|
|
base_query += " WHERE " + " AND ".join(where_clauses)
|
|
|
|
# --- Count ---
|
|
total_query = f"SELECT COUNT(*) FROM ({base_query}) AS total"
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
cur.execute(total_query, params)
|
|
total_count = cur.fetchone()[0]
|
|
|
|
# --- Sort + Paginate ---
|
|
sort_column_map = {
|
|
"id": "id",
|
|
"key": "key",
|
|
"value": "value",
|
|
"type": "type",
|
|
"created_at": "created_at",
|
|
"updated_at": "updated_at"
|
|
}
|
|
sort_col = sort_column_map.get(sort_by, "id")
|
|
sort_order = "DESC" if sort_order.upper() == "DESC" else "ASC"
|
|
|
|
base_query += f" ORDER BY {sort_col} {sort_order} LIMIT ? OFFSET ?"
|
|
params.extend([limit, offset])
|
|
|
|
cur.execute(base_query, params)
|
|
rows = cur.fetchall()
|
|
|
|
items = [
|
|
{
|
|
"id": r[0],
|
|
"key": r[1],
|
|
"value": r[2],
|
|
"type": r[3], # <-- thêm type
|
|
"created_at": r[4],
|
|
"updated_at": r[5]
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
conn.close()
|
|
return items, total_count
|
|
|
|
|
|
@staticmethod
|
|
def get(key: str, default=None):
|
|
"""Lấy giá trị setting theo key."""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT value FROM settings WHERE key = ?", (key,))
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
return row["value"] if row else default
|
|
|
|
@classmethod
|
|
def update_value(cls, setting_id, new_value):
|
|
s = cls.get_by_id(setting_id)
|
|
if not s:
|
|
raise ValueError(f"Setting id={setting_id} not found")
|
|
s["value"] = new_value
|
|
# Lưu vào DB
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("UPDATE settings SET value=? WHERE id=?", (new_value, setting_id))
|
|
conn.commit()
|
|
conn.close()
|
|
return True
|
|
|
|
|