Typing support and sqlite3.Row!
Typing will be handy in further development, as well will i settle down to sqlite3.Row. Fetchall() returns now a list(grr) with dict-imitating/sumating objects, which you can call keys() on and will return them, what we now already use for Backups, which comes handy. Also Typing gives the ability to let the code even more documentate itself. It's planned to use all over the place but i've to read myself into it yet more to get started on that one. It's a step in the right direction!
This commit is contained in:
parent
84508820a6
commit
8d4d2038d3
3 changed files with 29 additions and 18 deletions
|
@ -96,7 +96,10 @@ class Backup:
|
|||
returner = io.StringIO()
|
||||
write_csv = csv.DictWriter(returner, fieldnames=self.field_names, quoting=self.quoting, dialect=self.dialect)
|
||||
write_csv.writeheader()
|
||||
write_csv.writerows(fetched)
|
||||
for row in fetched:
|
||||
write_csv.writerow(dict(row))
|
||||
# sqlite3.Row doesn't "easily" convert to a dict itself sadly, so just a quick help from us here
|
||||
# it actually even delivers a list(sqlite3.Row) also, which doesnt make the life a whole lot easier
|
||||
|
||||
if self.filename == "stdout":
|
||||
print(returner.getvalue())
|
||||
|
@ -116,8 +119,13 @@ if __name__ == "__main__":
|
|||
L = ListUsers.ListUsers(config['DEFAULT']['applications_db'],
|
||||
unapproved=args.unapproved, approved=args.approved)
|
||||
fetch = L.get_fetch()
|
||||
B = Backup(args.file)
|
||||
B.backup_to_file(fetch)
|
||||
if fetch:
|
||||
B = Backup(args.file)
|
||||
B.setFieldnames(fetch[0].keys()) # sqlite3.row delivers its keys for us! SO NICE!
|
||||
B.backup_to_file(fetch)
|
||||
else:
|
||||
print("nothing to backup!")
|
||||
exit(1)
|
||||
exit(0)
|
||||
except KeyboardInterrupt as e:
|
||||
pass
|
||||
|
|
|
@ -1,9 +1,12 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from lib.sqlitedb import SQLiteDB
|
||||
import configparser
|
||||
import lib.uis.default as default_cmd # Follows -u, -a, -f flags
|
||||
|
||||
from typing import List # Typing support!
|
||||
import sqlite3 # sqlite3.Row-Object
|
||||
import configparser
|
||||
|
||||
|
||||
class ListUsers:
|
||||
db = None
|
||||
|
@ -25,13 +28,12 @@ class ListUsers:
|
|||
query = "SELECT * FROM `applications` WHERE `status` = '0'"
|
||||
elif approved: # Approved users
|
||||
query = "SELECT * FROM `applications` WHERE `status` = '1'"
|
||||
elif single_user is not None:
|
||||
query = "SELECT * FROM `applications` WHERE `username` = ?"
|
||||
self.usersFetch = self.db.safequery(query, tuple([single_user]))
|
||||
return
|
||||
else: # All users
|
||||
query = "SELECT * FROM `applications`"
|
||||
self.usersFetch = self.db.query(query)
|
||||
if single_user is not None:
|
||||
query = "SELECT * FROM `applications` WHERE `username` = ?"
|
||||
self.usersFetch = self.db.safequery(query, tuple([single_user]))
|
||||
|
||||
def output_as_list(self) -> str:
|
||||
"""Generates a string with one (approved) user per line and one newline at the end
|
||||
|
@ -50,11 +52,11 @@ class ListUsers:
|
|||
def prettyPrint(self) -> None:
|
||||
pass # see below why not implemented yet, texttable...
|
||||
|
||||
def get_fetch(self) -> list:
|
||||
def get_fetch(self) -> List[sqlite3.Row]:
|
||||
""" Returns a complete fetch done by the lib.sqlitedb-class
|
||||
|
||||
:return: Complete fetchall() in a dict
|
||||
:rtype: list
|
||||
:return: Complete fetchall(). A List[sqlite3.Row] with dict-emulation objects.
|
||||
:rtype: List[sqlite3.Row]
|
||||
"""
|
||||
|
||||
return self.usersFetch
|
||||
|
|
|
@ -1,11 +1,12 @@
|
|||
#!/usr/bin/env python3
|
||||
import sqlite3
|
||||
from sys import stderr as stderr
|
||||
from typing import List # Typing support!
|
||||
|
||||
|
||||
# create dictionary out of sqlite results
|
||||
def dict_factory(cursor, row):
|
||||
d = {}
|
||||
d: dict = {}
|
||||
for idx, col in enumerate(cursor.description):
|
||||
d[col[0]] = row[idx]
|
||||
return d
|
||||
|
@ -34,7 +35,7 @@ class SQLiteDB:
|
|||
except sqlite3.Error as e:
|
||||
print("Connection error: %s" % e, file=stderr)
|
||||
|
||||
self.cursor.row_factory = dict_factory # every result will be a dict now
|
||||
self.cursor.row_factory = sqlite3.Row # every result will be a dict now
|
||||
|
||||
def __del__(self):
|
||||
try:
|
||||
|
@ -43,7 +44,7 @@ class SQLiteDB:
|
|||
except sqlite3.Error as e:
|
||||
print("Couldn't gracefully close db: %s" % e, file=stderr)
|
||||
|
||||
def query(self, qq: str) -> list:
|
||||
def query(self, qq: str) -> List[sqlite3.Row]:
|
||||
"""Do a query and automagically get the fetched results in a list
|
||||
:param qq: Query to execute
|
||||
:type qq: str
|
||||
|
@ -71,14 +72,14 @@ class SQLiteDB:
|
|||
|
||||
# we could try to utilise that ourselfs in a function. Be c a r e f u l, these values in the tuple MUST HAVE
|
||||
# THE RIGHT TYPE
|
||||
def safequery(self, qq: str, deliver: tuple) -> list:
|
||||
def safequery(self, qq: str, deliver: tuple) -> List[sqlite3.Row]:
|
||||
""" Shall handle any query that has user input in it as an alternative to self.query
|
||||
:param qq: Query to execute
|
||||
:type qq: str
|
||||
:param deliver: User inputs marked with the placeholder(`?`) in the str
|
||||
:type deliver: tuple
|
||||
:returns: A tuple(/list) consisting with any fetched results
|
||||
:rtype: list
|
||||
:rtype: List[sqlite3.Row]
|
||||
"""
|
||||
|
||||
try:
|
||||
|
@ -88,7 +89,7 @@ class SQLiteDB:
|
|||
except TypeError as e:
|
||||
print("Types in given tuple doesnt match to execute query \"%s\": %s" % (qq, e), file=stderr)
|
||||
self.last_result = []
|
||||
except sqlite3.OperationalError as e:
|
||||
except sqlite3.OperationalError:
|
||||
self._createTable()
|
||||
return self.safequery(qq, deliver)
|
||||
except sqlite3.Error as e:
|
||||
|
@ -138,7 +139,7 @@ class SQLiteDB:
|
|||
return False
|
||||
return True
|
||||
|
||||
def _createTable(self):
|
||||
def _createTable(self) -> None:
|
||||
try:
|
||||
self.cursor.execute(
|
||||
"CREATE TABLE IF NOT EXISTS applications("
|
||||
|
|
Loading…
Reference in a new issue