restructure api key handling

 * include possibility of calling with API key value to
   better enable local testing of workflows

Author: JunAishima

data_validation.py

@@ -1,17 +1,11 @@
-import os
-
-from dotenv import load_dotenv
 from prefect import task, flow, get_run_logger
 import time as ttime
 from tiled.client import from_profile
 
 
 @task(retries=2, retry_delay_seconds=10)
-def read_all_streams(uid, beamline_acronym):
+def read_all_streams(uid, beamline_acronym, api_key=None):
     logger = get_run_logger()
-    with open("/srv/tiled.secret", "r") as secrets:
-        load_dotenv(stream=secrets)
-    api_key = os.environ["TILED_API_KEY"]
     cl = from_profile("nsls2", api_key=api_key)
     run = cl["tst"]["raw"][uid]
     logger.info(f"Validating uid {run.start['uid']}")
@@ -28,5 +22,5 @@ def read_all_streams(uid, beamline_acronym):
 
 
 @flow
-def data_validation(uid):
-    read_all_streams(uid, beamline_acronym="tst")
+def data_validation(uid, api_key=None):
+    read_all_streams(uid, beamline_acronym="tst", api_key=api_key)

end_of_run_workflow.py

@@ -1,9 +1,23 @@
+import os
+
 from prefect import task, flow, get_run_logger
 from data_validation import data_validation
 from test_extra_client import get_other_docs
+from dotenv import load_dotenv
 # from long_flow import long_flow
 
 
+def get_api_key_from_env(api_key=None):
+    logger = get_run_logger()
+    try:
+        with open("/srv/tiled.secret", "r") as secrets:
+            load_dotenv(stream=secrets)
+        api_key = os.environ["TILED_API_KEY"]
+    except Exception:
+        logger.exception("Exception while getting Tiled API key")
+    return api_key
+
+
 @task
 def log_completion():
     logger = get_run_logger()
@@ -14,7 +28,8 @@ def log_completion():
 def end_of_run_workflow(stop_doc):
     uid = stop_doc["run_start"]
     # hello_world()
-    data_validation(uid, return_state=True)
-    get_other_docs(uid)
+    api_key = get_api_key_from_env(api_key=None)
+    data_validation(uid, return_state=True, api_key=api_key)
+    get_other_docs(uid, api_key=api_key)
     # long_flow(iterations=100, sleep_length=10)
     log_completion()