Merge pull request #31 from cityofaustin/charlie/new-task-order-fields

new fields for Task Orders

Author: Charlie-Henry

.dockerignore

@@ -1,3 +1,13 @@
-env_file
-.git
-*.env
+# Ignores everything
+*
+
+# Only allows the following files:
+!config.py
+!fdu_program_tagging.py
+!prefix_maps.py
+!queries.py
+!s3_to_knack.py
+!s3_to_socrata.py
+!upload_to_s3.py
+!utils.py
+!requirements.txt

README.md

@@ -75,3 +75,10 @@ Required environmental variables, which are available in the DTS credential stor
 - `AWS_SECRET_ACCESS_KEY`: The secret key for your AWS account
 - `KNACK_APP_ID`: The Knack app ID of the destiantion knack app
 - `KNACK_API_KEY`: The kanck API key of the destination knack app
+
+You can also pass a `-p` or `--progress-bar` flag to see a progress bar while the script is running. This can be helpful
+if you are running this script on a large dataset for the first time.
+
+```shell
+$ python s3_to_knack.py task_orders data-tracker -p
+```

config.py

@@ -112,6 +112,18 @@ def stringify_value(value):
                 "finance-purchasing": "field_997",
                 "handler": add_comma_separator,
             },
+            {
+                "src": "LN_AMOUNT",
+                "data-tracker": "field_5365",
+                "finance-purchasing": "field_1350",
+                "handler": add_comma_separator,
+            },
+            {
+                "src": "TASK_ORDER_REM_AMT",
+                "data-tracker": "field_5366",
+                "finance-purchasing": "field_1351",
+                "handler": add_comma_separator,
+            },
             {
                 "src": "TASK_ORDER_ESTIMATOR",
                 "data-tracker": "field_4495",

env_template

@@ -0,0 +1,27 @@
+# AWS S3 (1pass: atd-finance-data)
+AWS_ACCESS_KEY_ID=
+AWS_SECRET_ACCESS_KEY=
+BUCKET=
+
+# Knack API (depends on which Knack app you want to update.)
+KNACK_API_KEY=
+KNACK_APP_ID=
+
+# Finance Data Warehouse Oracle DB (1pass: atd-finance-data)
+HOST=
+PASSWORD=
+PORT=
+SERVICE=
+USER=
+
+# 1pass: Socrata Key ID, Secret, and Token
+SO_USER=<apiKeyId>
+SO_PASS=<apiKeySecret>
+SO_TOKEN=<appToken>
+SO_WEB=datahub.austintexas.gov
+
+# Socrata Dataset Resource IDs
+TASK_DATASET=ggav-ufvc
+FDU_DATASET=jega-nqf6
+DEPT_UNITS_DATASET=bgrt-2m2z
+SUBPROJECTS_DATASET=x454-x7cu

queries.py

@@ -18,6 +18,8 @@
         atd_tk.CHARGED_AMOUNT,
         atd_tk.TASK_ORDER_BAL,
         atd_tk.TASK_ORDER_ESTIMATOR,
+        atd_tk.LN_AMOUNT,
+        atd_tk.TASK_ORDER_REM_AMT,
         buyer_tk.BYR_FDU
     FROM
         DEPT_2400_TK_VW atd_tk

requirements.txt

@@ -3,3 +3,4 @@ knackpy==1.0.*
 sodapy==2.1.*
 boto3==1.19.*
 oracledb==2.1.*
+tqdm==4.67.*

s3_to_knack.py

@@ -11,6 +11,7 @@
 
 import boto3
 import knackpy
+from tqdm import tqdm
 
 from config import FIELD_MAPS
 
@@ -35,6 +36,17 @@ def cli_args():
         choices=["data-tracker", "finance-purchasing"],
         help="The name of the destination Knack app",
     )
+    parser.add_argument(
+        "--progress-bar",
+        "-p",
+        action="store_true",
+        help=(
+            "Display a tqdm progress bar instead of the default periodic log "
+            "messages. Not recommended when running under Airflow, since the "
+            "bar's carriage-return redraws don't render cleanly in Airflow's "
+            "log viewer."
+        ),
+    )
     return parser.parse_args()
 
 
@@ -182,10 +194,10 @@ def coalesce_records(records_current, coalesce_fields, current_pk, separator=",\
     return list(index.values())
 
 
-def process_record(record, knack_obj, app, count):
+def process_record(record, knack_obj, app, count=None):
     method = "create" if not record.get("id") else "update"
     app.record(data=record, method=method, obj=knack_obj)
-    if count % 10 == 0:
+    if count is not None and count % 10 == 0:
         logging.info(f"{count} records processed so far...")
 
 
@@ -234,17 +246,31 @@ def main():
     logging.info(f"{len(todos)} records to process.")
 
     # Process todos with 10 workers
-    count = 0
     with ThreadPoolExecutor(max_workers=10) as executor:
-        futures = []
-        for record in todos:
-            futures.append(
-                executor.submit(process_record, record, knack_obj, app, count)
-            )
-            count += 1
-
-    for future in futures:
-        future.result()  # This will re-raise any exceptions encountered in the threads
+        if args.progress_bar:
+            # tqdm progress bar (best for interactive/terminal use; the
+            # carriage-return redraws don't render cleanly in Airflow logs)
+            futures = [
+                executor.submit(process_record, record, knack_obj, app)
+                for record in todos
+            ]
+            for future in tqdm(
+                futures, total=len(futures), desc="Processing records", unit="rec"
+            ):
+                future.result()  # This will re-raise any exceptions encountered in the threads
+        else:
+            # Default: periodic log messages, which render cleanly in Airflow's
+            # log viewer
+            futures = []
+            count = 0
+            for record in todos:
+                futures.append(
+                    executor.submit(process_record, record, knack_obj, app, count)
+                )
+                count += 1
+
+            for future in futures:
+                future.result()  # This will re-raise any exceptions encountered in the threads
 
 
 if __name__ == "__main__":

s3_to_socrata.py

@@ -153,6 +153,8 @@ def transform_tks(data):
         "CHARGED_AMOUNT": "CHARGEDAMOUNT",
         "TASK_ORDER_BAL": "BALANCE",
         "BYR_FDU": "BUYER_FDUS",
+        "LN_AMOUNT": "LN_AMOUNT",
+        "TASK_ORDER_REM_AMT": "TASK_ORDER_REM_AMT",
     }
 
     replaced_data = []