xan/xenu_nntp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Database._sqlite3_function() for convenience

Browse source
This commit is contained in:
XANTRONIX Development 2024-11-09 12:40:51 -05:00
parent 2d66000b7d
commit c260909ab4
1 changed files with 12 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
lib/nntp/tiny/db.py
View file

@ -117,28 +117,22 @@ class Database():
def get(self, table, values: dict=dict()): def get(self, table, values: dict=dict()):
return self.query(table, values).fetchone() return self.query(table, values).fetchone()
def min(self, table, column: str) -> int: def _sqlite3_function(self, table, fn: str, column: str, values: dict=dict()) -> int:
sql = f"select min({column}) as _min from {table.name}" sql = f"select {fn}({column}) as ret from {table.name}"
row = self.db.execute(sql).fetchone()
return row[0] if row is not None else None
def max(self, table, column: str) -> int:
sql = f"select max({column}) as _max from {table.name}"
row = self.db.execute(sql).fetchone()
return row[0] if row is not None else None
def count(self, table, values: dict=dict()):
sql = f"select count(id) as num from {table.name}"
if len(values) > 0: if len(values) > 0:
sql += " where " sql += " where "
sql += " and ".join([f"{k} = ?" for k in values]) sql += " and ".join([f"{k} = ?" for k in values])
cr = self.db.execute(sql, list(values.values())) row = self.db.execute(sql, list(values.values())).fetchone()
row = cr.fetchone()
return row[0] if row is not None else None return row[0] if row is not None else None
def min(self, table, column: str, values: dict=dict()) -> int:
return self._sqlite3_function(table, 'min', column, values)
def max(self, table, column: str, values: dict=dict()) -> int:
return self._sqlite3_function(table, 'max', column, values)
def count(self, table, values: dict=dict()) -> int:
return self._sqlite3_function(table, 'count', table.key, values)