Browse Source

move common display mapper to db_utils and add type hints

Daniel Sheffield 3 years ago
parent
commit
a6c5e95aee
3 changed files with 36 additions and 37 deletions
  1. 15 1
      app/db_utils.py
  2. 11 21
      grocery_transactions.py
  3. 10 15
      price_check.py

+ 15 - 1
app/db_utils.py

@@ -4,6 +4,8 @@
 # All rights reserved
 #
 # THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
+import time
+from typing import Any, Callable
 from .txn_view import (
     get_table_statement,
     get_transactions_statement,
@@ -23,6 +25,16 @@ NON_IDENTIFIER_COLUMNS = [
     'organic',
 ]
 
+display_map = {
+    'ts': lambda x: f"{time.strftime('%Y-%m-%d %H:%M', (x.year, x.month, x.day, x.hour, x.minute, 0, 0, 0, 0))}",
+    'price': lambda x: f'{x:.4f}',
+    'quantity': lambda x: f'{x:.2f}',
+    'organic': lambda x: 'yes' if x else 'no',
+}
+display_mapper: Callable[
+    [Any, str], str
+] = lambda data, name: display_map[name](data) if name in display_map else data
+
 def cursor_as_dict(cur):
     _col_idx_map=dict(map(lambda col: (col[1].name, col[0]), enumerate(cur.description)))
     for row in map(lambda row, _map=_col_idx_map: dict([
@@ -103,7 +115,9 @@ def get_insert_product_statement(product, category, group):
 
 class QueryManager(object):
     
-    def __init__(self, cursor, display):
+    def __init__(self, cursor, display: Callable[
+        [Any, str], str
+    ]):
         self.display = display
         self.cursor = cursor
     

+ 11 - 21
grocery_transactions.py

@@ -5,22 +5,19 @@
 # All rights reserved
 #
 # THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
-import time
 import itertools
 import sys
+from sqlite3 import Cursor
 from typing import Union
+
 import urwid
 from urwid import raw_display
-from app.db_utils import (
-    QueryManager,
-    get_insert_product_statement
-)
-from app.activities import (
-    show_or_exit,
-    ActivityManager,
-)  
-from app.activities.TransactionEditor import TransactionEditor
+
+from app.activities import ActivityManager, show_or_exit
 from app.activities.NewProduct import NewProduct
+from app.activities.TransactionEditor import TransactionEditor
+from app.db_utils import (QueryManager, display_mapper,
+                          get_insert_product_statement)
 from app.widgets import AutoCompleteEdit, AutoCompleteFloatEdit
 
 try:
@@ -32,10 +29,11 @@ except:
     password = ''
 
 try:
-    import psycopg2
     import os
+
+    import psycopg2
     user = os.getenv('USER')
-    conn = psycopg2.connect(f"{host} dbname=das user={user} {password}")
+    conn: Cursor = psycopg2.connect(f"{host} dbname=das user={user} {password}")
     cur = conn.cursor()
 except:
     print('Failed to set up db connection. Entering Mock mode')
@@ -81,14 +79,6 @@ cols = [
     )
 ]
 
-display_map = {
-    'ts': lambda x: f"{time.strftime('%Y-%m-%d %H:%M', (x.year, x.month, x.day, x.hour, x.minute, 0, 0, 0, 0))}",
-    'price': lambda x: f"{x:.2f}",
-    'quantity': lambda x: f"{x:.2f}",
-    'organic': lambda x: "true" if x else "false",
-}
-display = lambda data, name: display_map[name](data) if name in display_map else data
-
 def _apply_choice_callback(
     activity_manager: ActivityManager,
     base: str, name: str, value: str
@@ -213,7 +203,7 @@ class GroceryTransactionEditor(urwid.WidgetPlaceholder):
 cur.execute("BEGIN")
 
 activity_manager = ActivityManager()
-query_manager = QueryManager(cur, display)
+query_manager = QueryManager(cur, display_mapper)
 
 activity_manager.create(TransactionEditor, 'transaction',
     query_manager, cols, grid_layout, side_pane, bottom_pane,

+ 10 - 15
price_check.py

@@ -6,16 +6,17 @@
 #
 # THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
 import itertools
+from sqlite3 import Cursor
 from typing import Union
+
 import urwid
 from urwid import raw_display
-from app.db_utils import QueryManager
-from app.activities import (
-    show_or_exit,
-    ActivityManager,
-)
+
+from app.activities import ActivityManager, show_or_exit
 from app.activities.PriceCheck import PriceCheck
+from app.db_utils import QueryManager, display_mapper
 from app.widgets import AutoCompleteEdit, AutoCompleteFloatEdit
+
 try:
     from db_credentials import HOST, PASSWORD
     host = f'host={HOST}'
@@ -25,11 +26,12 @@ except:
     password = ''
 
 try:
-    import psycopg2
     import os
+
+    import psycopg2
     user = os.getenv('USER')
     conn = psycopg2.connect(f"{host} dbname=das user={user} {password}")
-    cur = conn.cursor()
+    cur: Cursor = conn.cursor()
 except:
     print('Failed to set up db connection. Entering Mock mode')
     exit(1)
@@ -84,13 +86,6 @@ inputs = filter(
     )
 )
 
-display_map = {
-    'price': lambda x: f'{x:.4f}',
-    'quantity': lambda x: f'{x:.2f}',
-    'organic': lambda x: 'true' if x else 'false',
-}
-display = lambda data, name: display_map[name](data) if name in display_map else data
-
 def _apply_choice_callback(
     activity_manager: ActivityManager,
     base: str, name: str, value: str
@@ -120,7 +115,7 @@ class GroceryPriceCheck(urwid.WidgetPlaceholder):
 cur.execute("BEGIN")
 
 activity_manager = ActivityManager()
-query_manager = QueryManager(cur, display)
+query_manager = QueryManager(cur, display_mapper)
 
 activity_manager.create(PriceCheck, 'price_check',
     query_manager, inputs, top_pane, left_pane, right_pane, bottom_pane, badge,