Tries to fix some linting issues in main

src/rxn_network/jobs/core.py

@@ -129,7 +129,7 @@ def make(self, chemsys: str):
             )
         else:
             try:
-                from mp_api.client import MPRester
+                from mp_api.client import MPRester  # noqa: PLC0415
             except ImportError as err:
                 raise ImportError("You may need to install the Materials Project API: pip install -U mp-api") from err
 

src/rxn_network/jobs/schema.py

@@ -1,6 +1,6 @@
 """Core definition for various task and synthesis recipe documents."""
 
-from typing import Any, Optional
+from typing import Any
 
 from pydantic import BaseModel, Field
 from pymatgen.core.composition import Element
@@ -18,38 +18,38 @@
 class EntrySetDocument(BaseModel):
     """A single entry set object as produced by the GetEntrySet job."""
 
-    task_label: Optional[str] = Field(None, description="The name of the task.")
+    task_label: str | None = Field(None, description="The name of the task.")
     last_updated: str = Field(
         default_factory=datetime_str,
         description="Timestamp of when the document was last updated.",
     )
     entries: GibbsEntrySet = Field(description="The entry set object.")
-    e_above_hull: Optional[float] = Field(None, description="The e_above_hull cutoff.")
+    e_above_hull: float | None = Field(None, description="The e_above_hull cutoff.")
     include_polymorphs: bool = Field(
         default=False, description="Whether to include metastable polymorphs in the entry set."
     )
-    formulas_to_include: Optional[list[str]] = Field(
+    formulas_to_include: list[str] | None = Field(
         None, description="The required formulas to include during construciton."
     )
 
 
 class EnumeratorTaskDocument(BaseModel):
     """A single task object from the reaction enumerator workflow."""
 
-    task_label: Optional[str] = Field(None, description="The name of the task.")
+    task_label: str | None = Field(None, description="The name of the task.")
     last_updated: str = Field(
         default_factory=datetime_str,
         description="Timestamp of when the document was last updated.",
     )
     rxns: ReactionSet = Field(description="The reaction set.")
-    targets: Optional[list[str]] = Field(None, description="The target formulas used in the enumerator(s).")
-    elements: Optional[list[Element]] = Field(None, description="The elements of the total chemical system")
-    chemsys: Optional[str] = Field(None, description="The total chemical system string (e.g., Fe-Li-O).")
-    added_elements: Optional[list[Element]] = Field(
+    targets: list[str] | None = Field(None, description="The target formulas used in the enumerator(s).")
+    elements: list[Element] | None = Field(None, description="The elements of the total chemical system")
+    chemsys: str | None = Field(None, description="The total chemical system string (e.g., Fe-Li-O).")
+    added_elements: list[Element] | None = Field(
         None, description="The elements added beyond the elements of the target(s)."
     )
-    added_chemsys: Optional[str] = Field(None, description="The chemical system of the added elements")
-    enumerators: Optional[list[Enumerator]] = Field(
+    added_chemsys: str | None = Field(None, description="The chemical system of the added elements")
+    enumerators: list[Enumerator] | None = Field(
         None,
         description="A list of the enumerator objects used to calculate the reactions.",
     )
@@ -60,7 +60,7 @@ class CompetitionTaskDocument(BaseModel):
     CalculateCompetitionMaker job.
     """
 
-    task_label: Optional[str] = Field(None, description="The name of the task.")
+    task_label: str | None = Field(None, description="The name of the task.")
     last_updated: str = Field(
         default_factory=datetime_str,
         description="Timestamp of when the document was last updated.",
@@ -71,26 +71,26 @@ class CompetitionTaskDocument(BaseModel):
             " information stored in their data attribute."
         )
     )
-    precursor_formula: Optional[str] = Field(
+    precursor_formula: str | None = Field(
         None, description="The reduced chemical formula of the precursor material (used in extraction flows)."
     )
-    target_formula: Optional[str] = Field(None, description="The reduced chemical formula of the target material.")
-    open_elem: Optional[Element] = Field(None, description="The open element (if any).")
-    chempot: Optional[float] = Field(None, description="The chemical potential of the open element.")
-    added_elements: Optional[list[Element]] = Field(
+    target_formula: str | None = Field(None, description="The reduced chemical formula of the target material.")
+    open_elem: Element | None = Field(None, description="The open element (if any).")
+    chempot: float | None = Field(None, description="The chemical potential of the open element.")
+    added_elements: list[Element] | None = Field(
         None, description="The elements added beyond the elements of the target(s)."
     )
-    added_chemsys: Optional[str] = Field(None, description="The chemical system of the added elements.")
-    calculate_competition: Optional[bool] = Field(None, description="Whether to calculate competition scores.")
-    calculate_chempot_distances: Optional[bool] = Field(
+    added_chemsys: str | None = Field(None, description="The chemical system of the added elements.")
+    calculate_competition: bool | None = Field(None, description="Whether to calculate competition scores.")
+    calculate_chempot_distances: bool | None = Field(
         None, description="Whether to calculate chemical potential distances."
     )
-    temp: Optional[float] = Field(
+    temp: float | None = Field(
         None,
         description=("The temperature in K used to determine the primary competition weightings."),
     )
-    batch_size: Optional[int] = Field(None, description="The batch size for the reaction set")
-    cpd_kwargs: Optional[dict[str, Any]] = Field(None, description="The kwargs for ChempotDistanceCalculator.")
+    batch_size: int | None = Field(None, description="The batch size for the reaction set")
+    cpd_kwargs: dict[str, Any] | None = Field(None, description="The kwargs for ChempotDistanceCalculator.")
 
 
 class NetworkTaskDocument(BaseModel):
@@ -99,24 +99,24 @@ class NetworkTaskDocument(BaseModel):
     Optionally includes unbalanced paths found during pathfinding.
     """
 
-    task_label: Optional[str] = Field(None, description="The name of the task.")
+    task_label: str | None = Field(None, description="The name of the task.")
     last_updated: str = Field(
         default_factory=datetime_str,
         description="Timestamp of when the document was last updated.",
     )
     network: Network = Field(description="The reaction network")
-    paths: Optional[PathwaySet] = Field(None, description="The (simple/unbalanced) reaction pathways")
-    k: Optional[int] = Field(None, description="The number of paths solved for, if any.")
-    precursors: Optional[list[str]] = Field(None, description="The precursor formulas.")
-    targets: Optional[list[str]] = Field(None, description="The target formulas.")
+    paths: PathwaySet | None = Field(None, description="The (simple/unbalanced) reaction pathways")
+    k: int | None = Field(None, description="The number of paths solved for, if any.")
+    precursors: list[str] | None = Field(None, description="The precursor formulas.")
+    targets: list[str] | None = Field(None, description="The target formulas.")
 
 
 class PathwaySolverTaskDocument(BaseModel):
     """The calculation output from the PathwaySolverMaker workflow. Contains the pathway
     solver object and its calculated balanced pathways.
     """
 
-    task_label: Optional[str] = Field(None, description="The name of the task.")
+    task_label: str | None = Field(None, description="The name of the task.")
     last_updated: str = Field(
         default_factory=datetime_str,
         description="Timestamp of when the document was last updated.",

src/rxn_network/utils/ray.py

@@ -27,7 +27,7 @@ def initialize_ray(quiet: bool = False):
         logger.setLevel("INFO")
     if not ray.is_initialized():
         logger.info("Ray is not initialized. Checking for existing cluster...")
-        if os.environ.get("IP_HEAD") or int(os.environ.get("PBS_NNODES", 0)) > 1:
+        if os.environ.get("IP_HEAD") or int(os.environ.get("PBS_NNODES", "0")) > 1:
             ray.init(
                 address="auto",
             )

tests/utils/test_ray.py

@@ -51,7 +51,7 @@ def test_initialize_ray_with_slurm_cluster(capsys):
     try:
         initialize_ray()
     except Exception as error:  # if it does fail, make sure it's the right error
-        assert "ConnectionError" in str(error.type)  # noqa
+        assert "ConnectionError" in type(error).__name__
 
     if "Connected" in capsys.readouterr().err:
         pytest.skip("Skipping test_initialize_ray_with_slurm_cluster due to existing cluster")
@@ -67,7 +67,7 @@ def test_initialize_ray_with_pbs_cluster(capsys):
     try:
         initialize_ray()
     except Exception as error:  # if it does fail, make sure it's the right error
-        assert "ConnectionError" in str(error.type)  # noqa
+        assert "ConnectionError" in type(error).__name__
 
     if "Connected" in capsys.readouterr().err:
         pytest.skip("Skipping test_initialize_ray_with_slurm_cluster due to existing cluster")