use self-documenting f-strings

fireworks/core/firework.py

@@ -471,7 +471,7 @@ def __init__(
             fw_id (int): id of the Firework this Launch is running.
         """
         if state not in Firework.STATE_RANKS:
-            raise ValueError(f"Invalid launch state: {state}")
+            raise ValueError(f"Invalid launch {state=}")
         self.launch_dir = launch_dir
         self.fworker = fworker or FWorker()
         self.host = host or get_my_host()
@@ -1029,7 +1029,7 @@ def append_wf(self, new_wf, fw_ids, detour=False, pull_spec_mods=False):
                     ready_run = [(f >= 0 and Firework.STATE_RANKS[self.fw_states[f]] > 1) for f in self.links[fw_id]]
                     if any(ready_run):
                         raise ValueError(
-                            f"fw_id: {fw_id}: Detour option only works if all children "
+                            f"{fw_id=}: Detour option only works if all children "
                             "of detours are not READY to run and have not already run"
                         )
 

fireworks/core/launchpad.py

@@ -460,7 +460,7 @@ def get_launch_by_id(self, launch_id):
         if m_launch:
             m_launch["action"] = get_action_from_gridfs(m_launch.get("action"), self.gridfs_fallback)
             return Launch.from_dict(m_launch)
-        raise ValueError(f"No Launch exists with launch_id: {launch_id}")
+        raise ValueError(f"No Launch exists with {launch_id=}")
 
     def get_fw_dict_by_id(self, fw_id):
         """
@@ -513,7 +513,7 @@ def get_wf_by_fw_id(self, fw_id):
         """
         links_dict = self.workflows.find_one({"nodes": fw_id})
         if not links_dict:
-            raise ValueError(f"Could not find a Workflow with fw_id: {fw_id}")
+            raise ValueError(f"Could not find a Workflow with {fw_id=}")
         fws = map(self.get_fw_by_id, links_dict["nodes"])
         return Workflow(
             fws,
@@ -535,7 +535,7 @@ def get_wf_by_fw_id_lzyfw(self, fw_id: int) -> Workflow:
         """
         links_dict = self.workflows.find_one({"nodes": fw_id})
         if not links_dict:
-            raise ValueError(f"Could not find a Workflow with fw_id: {fw_id}")
+            raise ValueError(f"Could not find a Workflow with {fw_id=}")
 
         fws = [
             LazyFirework(fw_id, self.fireworks, self.launches, self.gridfs_fallback) for fw_id in links_dict["nodes"]
@@ -961,7 +961,7 @@ def pause_fw(self, fw_id):
         if f:
             self._refresh_wf(fw_id)
         if not f:
-            self.m_logger.error(f"No pausable (WAITING,READY,RESERVED) Firework exists with fw_id: {fw_id}")
+            self.m_logger.error(f"No pausable (WAITING,READY,RESERVED) Firework exists with {fw_id=}")
         return f
 
     def defuse_fw(self, fw_id, rerun_duplicates=True):
@@ -1428,7 +1428,7 @@ def checkout_fw(self, fworker, launch_dir, fw_id=None, host=None, ip=None, state
         # insert the launch
         self.launches.find_one_and_replace({"launch_id": m_launch.launch_id}, m_launch.to_db_dict(), upsert=True)
 
-        self.m_logger.debug(f"Created/updated Launch with launch_id: {launch_id}")
+        self.m_logger.debug(f"Created/updated Launch with {launch_id=}")
 
         # update the firework's launches
         if not reserved_launch:
@@ -1673,9 +1673,9 @@ def rerun_fw(self, fw_id, rerun_duplicates=True, recover_launch=None, recover_mo
 
         # rerun this FW
         if m_fw["state"] in ["ARCHIVED", "DEFUSED"]:
-            self.m_logger.info(f"Cannot rerun fw_id: {fw_id}: it is {m_fw['state']}.")
+            self.m_logger.info(f"Cannot rerun {fw_id=}: it is {m_fw['state']}.")
         elif m_fw["state"] == "WAITING" and not recover_launch:
-            self.m_logger.debug(f"Skipping rerun fw_id: {fw_id}: it is already WAITING.")
+            self.m_logger.debug(f"Skipping rerun {fw_id=}: it is already WAITING.")
         else:
             with WFLock(self, fw_id):
                 wf = self.get_wf_by_fw_id_lzyfw(fw_id)

fireworks/flask_site/app.py

@@ -114,23 +114,22 @@ def home():
 
     # Newest Workflows table data
     wfs_shown = app.lp.workflows.find(_addq_WF({}), limit=PER_PAGE, sort=[("_id", DESCENDING)])
-    wf_info = []
-    for item in wfs_shown:
-        wf_info.append(
-            {
-                "id": item["nodes"][0],
-                "name": item["name"],
-                "state": item["state"],
-                "fireworks": list(
-                    app.lp.fireworks.find(
-                        {"fw_id": {"$in": item["nodes"]}},
-                        limit=PER_PAGE,
-                        sort=[("fw_id", DESCENDING)],
-                        projection=["state", "name", "fw_id"],
-                    )
-                ),
-            }
-        )
+    wf_info = [
+        {
+            "id": item["nodes"][0],
+            "name": item["name"],
+            "state": item["state"],
+            "fireworks": list(
+                app.lp.fireworks.find(
+                    {"fw_id": {"$in": item["nodes"]}},
+                    limit=PER_PAGE,
+                    sort=[("fw_id", DESCENDING)],
+                    projection=["state", "name", "fw_id"],
+                )
+            ),
+        }
+        for item in wfs_shown
+    ]
 
     PLOTTING = False
     try:
@@ -144,7 +143,7 @@ def home():
 @app.route("/fw/<int:fw_id>/details")
 @requires_auth
 def get_fw_details(fw_id):
-    # just fill out whatever attributse you want to see per step, then edit the handlebars template in
+    # just fill out whatever attributes you want to see per step, then edit the handlebars template in
     # wf_details.html
     # to control their display
     fw = app.lp.get_fw_dict_by_id(fw_id)
@@ -162,7 +161,7 @@ def fw_details(fw_id):
     try:
         int(fw_id)
     except Exception:
-        raise ValueError(f"Invalid fw_id: {fw_id}")
+        raise ValueError(f"Invalid {fw_id=}")
     fw = app.lp.get_fw_dict_by_id(fw_id)
     fw = json.loads(json.dumps(fw, default=DATETIME_HANDLER))  # formats ObjectIds
     return render_template("fw_details.html", **locals())

fireworks/scripts/lpad_run.py

@@ -174,7 +174,7 @@ def init_yaml(args: Namespace) -> None:
     print("Please supply the following configuration values")
     print("(press Enter if you want to accept the defaults)\n")
     for k, default, helptext in fields:
-        val = input(f"Enter {k} parameter. (default: {default}). {helptext}: ")
+        val = input(f"Enter {k} parameter. ({default=}). {helptext}: ")
         doc[k] = val or default
     if "port" in doc:
         doc["port"] = int(doc["port"])  # enforce the port as an int
@@ -627,7 +627,7 @@ def rerun_fws(args: Namespace) -> None:
         launch_ids = [None] * len(fw_ids)
     for fw_id, l_id in zip(fw_ids, launch_ids):
         lp.rerun_fw(int(fw_id), recover_launch=l_id, recover_mode=args.recover_mode)
-        lp.m_logger.debug(f"Processed fw_id: {fw_id}")
+        lp.m_logger.debug(f"Processed {fw_id=}")
     lp.m_logger.info(f"Finished setting {len(fw_ids)} FWs to rerun")
 
 
@@ -645,10 +645,10 @@ def refresh(args: Namespace) -> None:
 def unlock(args: Namespace) -> None:
     lp = get_lp(args)
     fw_ids = parse_helper(lp, args, wf_mode=True)
-    for f in fw_ids:
-        with WFLock(lp, f, expire_secs=0, kill=True):
-            lp.m_logger.warning(f"FORCIBLY RELEASING LOCK DUE TO USER COMMAND, WF: {f}")
-            lp.m_logger.debug(f"Processed Workflow with fw_id: {f}")
+    for fw_id in fw_ids:
+        with WFLock(lp, fw_id, expire_secs=0, kill=True):
+            lp.m_logger.warning(f"FORCIBLY RELEASING LOCK DUE TO USER COMMAND, WF: {fw_id}")
+            lp.m_logger.debug(f"Processed Workflow with {fw_id=}")
     lp.m_logger.info(f"Finished unlocking {len(fw_ids)} Workflows")
 
 
@@ -813,16 +813,16 @@ def track_fws(args: Namespace) -> None:
     include = args.include
     exclude = args.exclude
     first_print = True  # used to control newline
-    for f in fw_ids:
-        data = lp.get_tracker_data(f)
+    for fw_id in fw_ids:
+        data = lp.get_tracker_data(fw_id)
         output = []
-        for d in data:
-            for t in d["trackers"]:
-                if (not include or t.filename in include) and (not exclude or t.filename not in exclude):
-                    output.extend((f"## Launch id: {d['launch_id']}", str(t)))
+        for dct in data:
+            for tracker in dct["trackers"]:
+                if (not include or tracker.filename in include) and (not exclude or tracker.filename not in exclude):
+                    output.extend((f"## Launch id: {dct['launch_id']}", str(tracker)))
         if output:
-            name = lp.fireworks.find_one({"fw_id": f}, {"name": 1})["name"]
-            output.insert(0, f"# FW id: {f}, FW name: {name}")
+            name = lp.fireworks.find_one({"fw_id": fw_id}, {"name": 1})["name"]
+            output.insert(0, f"# FW id: {fw_id}, FW {name=}")
             if first_print:
                 first_print = False
             else:

fireworks/tests/mongo_tests.py

@@ -546,7 +546,7 @@ def test_append_wf(self) -> None:
         assert new_fw.spec["dummy2"] == [True]
 
         new_wf = Workflow([Firework([ModSpecTask()])])
-        with pytest.raises(ValueError):
+        with pytest.raises(ValueError, match="Cannot append to a FW that is not in the original Workflow"):
             self.lp.append_wf(new_wf, [4], detour=True)
 
     def test_append_wf_detour(self) -> None:

fireworks/utilities/filepad.py

@@ -135,7 +135,7 @@ def add_file(self, path, identifier=None, compress=True, metadata=None):
         if identifier is not None:
             _, doc = self.get_file(identifier)
             if doc is not None:
-                self.logger.warning(f"identifier: {identifier} exists. Skipping insertion")
+                self.logger.warning(f"{identifier=} exists. Skipping insertion")
                 return doc["gfs_id"], doc["identifier"]
 
         path = os.path.abspath(path)