Revert "ruff fixes"

This reverts commit d979a3f.

alab_management/sample_view/__init__.py

@@ -1,5 +1,7 @@
-"""Sample related things."""
+"""
+Sample related things.
+"""
 
-from .completed_sample_view import CompletedSampleView
 from .sample import Sample, SamplePosition, add_standalone_sample_position
 from .sample_view import SampleView
+from .completed_sample_view import CompletedSampleView

alab_management/sample_view/completed_sample_view.py

@@ -1,8 +1,8 @@
-from typing import Union
-
+from typing import cast, Union
 from bson import ObjectId
 
-from alab_management.utils.data_objects import get_collection, get_completed_collection
+from ..utils.data_objects import get_collection, get_completed_collection
+from .sample_view import SampleView
 
 
 class CompletedSampleView:
@@ -11,7 +11,9 @@ def __init__(self):
         self._completed_sample_collection = get_completed_collection("samples")
 
     def save_sample(self, sample_id: ObjectId):
-        """Saves a sample dictionary to the completed database. This should be copying a sample from the working database to the completed database."""
+        """
+        Saves a sample dictionary to the completed database. This should be copying a sample from the working database to the completed database.
+        """
         # if self.exists(sample_id):
         #     raise ValueError(
         #         f"Sample with id {sample_id} already exists in the completed database!"
@@ -25,22 +27,21 @@ def save_sample(self, sample_id: ObjectId):
                 f"Sample with id {sample_id} does not exist in the database!"
             )
         if self.exists(sample_id):
-            self._completed_sample_collection.update_one(
+            result = self._completed_sample_collection.update_one(
                 filter={"_id": ObjectId(sample_id)},
                 update={"$set": sample_dict},
                 upsert=True,
             )
         else:
-            self._completed_sample_collection.insert_one(sample_dict)
+            result = self._completed_sample_collection.insert_one(sample_dict)
 
     def exists(self, sample_id: Union[ObjectId, str]) -> bool:
-        """Check if a sample exists in the database.
+        """Check if a sample exists in the database
 
         Args:
             sample_id (ObjectId): id of the sample within sample collection
 
-        Returns
-        -------
+        Returns:
             bool: True if sample exists in the database
         """
         return (

alab_management/sample_view/sample.py

@@ -1,15 +1,17 @@
-"""The definition of the Sample and SamplePosition classes."""
+"""
+The definition of the Sample and SamplePosition classes.
+"""
 
 from dataclasses import dataclass, field
-from typing import Any, ClassVar, Dict, List, Optional
+from typing import Any, List, Optional, ClassVar, Dict, Type
 
 from bson import ObjectId
 
 
 @dataclass(frozen=True)
 class Sample:
     """
-    Basic sample object.
+    Basic sample object
 
     - ``sample_id``: the unique id of a sample in the database
     - ``task_id``: the unique id of a task that currently "owns" (is processing) this sample
@@ -29,7 +31,7 @@ class Sample:
 @dataclass(frozen=True)
 class SamplePosition:
     """
-    A sample position in the lab.
+    A sample position in the lab
 
     Sample position is a position in the lab that can hold sample,
     it is not a geographic coordinate in the lab, but a defined
@@ -56,7 +58,9 @@ def __post_init__(self):
 
 
 def add_standalone_sample_position(position: SamplePosition):
-    """Register a device instance."""
+    """
+    Register a device instance
+    """
     if not isinstance(position, SamplePosition):
         raise TypeError(
             f"The type of position should be SamplePosition, but user provided {type(position)}"
@@ -67,5 +71,7 @@ def add_standalone_sample_position(position: SamplePosition):
 
 
 def get_all_standalone_sample_positions() -> Dict[str, SamplePosition]:
-    """Get all the device names in the device registry."""
+    """
+    Get all the device names in the device registry
+    """
     return _standalone_sample_position_registry.copy()

alab_management/sample_view/sample_view.py

@@ -1,17 +1,18 @@
-"""A wrapper over the ``samples`` and ``sample_positions`` collections."""
+"""
+A wrapper over the ``samples`` and ``sample_positions`` collections.
+"""
 
 import re
 from datetime import datetime
 from enum import Enum, auto
-from typing import Any, Dict, List, Optional, Tuple, Union, cast
+from typing import Optional, List, Dict, Any, Tuple, cast, Union
 
 import pymongo
 from bson import ObjectId
 from pydantic import BaseModel, conint
 
-from alab_management.utils.data_objects import get_collection, get_lock
-
 from .sample import Sample, SamplePosition
+from ..utils.data_objects import get_collection, get_lock
 
 
 class SamplePositionRequest(BaseModel):
@@ -23,7 +24,7 @@ class SamplePositionRequest(BaseModel):
     """
 
     class Config:
-        """raise error when extra kwargs."""
+        """raise error when extra kwargs"""
 
         extra = "forbid"
 
@@ -47,7 +48,7 @@ def from_py_type(
 
 class SamplePositionStatus(Enum):
     """
-    The status of a sample position.
+    The status of a sample position
 
     - ``EMPTY``: the sample position is neither locked nor occupied
     - ``OCCUPIED``: there is a sample in the sample position
@@ -60,7 +61,9 @@ class SamplePositionStatus(Enum):
 
 
 class SampleView:
-    """Sample view manages the samples and their positions."""
+    """
+    Sample view manages the samples and their positions
+    """
 
     def __init__(self):
         self._sample_collection = get_collection("samples")
@@ -81,7 +84,7 @@ def add_sample_positions_to_db(
         parent_device_name: Optional[str] = None,
     ):
         """
-        Insert sample positions info to db, which includes position name and description.
+        Insert sample positions info to db, which includes position name and description
 
         If one sample position's name has already appeared in the database,
         we will just skip it.
@@ -122,7 +125,9 @@ def add_sample_positions_to_db(
                     self._sample_positions_collection.insert_one(new_entry)
 
     def clean_up_sample_position_collection(self):
-        """Drop the sample position collection."""
+        """
+        Drop the sample position collection
+        """
         self._sample_positions_collection.drop()
 
     def request_sample_positions(
@@ -131,7 +136,7 @@ def request_sample_positions(
         sample_positions: List[Union[SamplePositionRequest, str, Dict[str, Any]]],
     ) -> Optional[Dict[str, List[Dict[str, Any]]]]:
         """
-        Request a list of sample positions, this function will return until all the sample positions are available.
+        Request a list of sample positions, this function will return until all the sample positions are available
 
         Args:
             task_id: the task id that requests these resources
@@ -147,7 +152,7 @@ def request_sample_positions(
         ]
 
         if len(sample_positions_request) != len(
-            {sample_position.prefix for sample_position in sample_positions_request}
+            set(sample_position.prefix for sample_position in sample_positions_request)
         ):
             raise ValueError("Duplicated sample_positions in one request.")
 
@@ -178,7 +183,7 @@ def request_sample_positions(
 
     def get_sample_position(self, position: str) -> Optional[Dict[str, Any]]:
         """
-        Get the sample position entry in the database.
+        Get the sample position entry in the database
 
         if the position is not a valid position (not defined in the database), return None
         """
@@ -188,7 +193,7 @@ def get_sample_position_status(
         self, position: str
     ) -> Tuple[SamplePositionStatus, Optional[ObjectId]]:
         """
-        Get the status of a sample position.
+        Get the status of a sample position
 
         If there is a sample in the position, return OCCUPIED;
         else if the sample position is locked by a task, return LOCKED;
@@ -197,8 +202,7 @@ def get_sample_position_status(
         Args:
             position: the name of the sample position
 
-        Returns
-        -------
+        Returns:
             if the position is occupied by a sample, return OCCUPIED and the task id of the sample
             if the position is locked by a task, return LOCKED and the task id
             else, return EMPTY and None
@@ -217,11 +221,13 @@ def get_sample_position_status(
         return SamplePositionStatus.EMPTY, None
 
     def get_sample_position_parent_device(self, position: str) -> Optional[str]:
-        """Get the parent device of a sample position. If no parent device is defined, returns None. If the "position" query is a prefix, will look for a single parent device across all matched positions (ie a query for position="furnace_1/tray" will properly return "furnace_1" even if "furnace_1/tray/1" and "furnace_1/tray/2" are in the database _as long as "furnace_1" is the parent device of both!_)."""
+        """
+        Get the parent device of a sample position. If no parent device is defined, returns None. If the "position" query is a prefix, will look for a single parent device across all matched positions (ie a query for position="furnace_1/tray" will properly return "furnace_1" even if "furnace_1/tray/1" and "furnace_1/tray/2" are in the database _as long as "furnace_1" is the parent device of both!_).
+        """
         sample_positions = self._sample_positions_collection.find(
             {"name": {"$regex": f"^{position}"}}
         )
-        parent_devices = list({sp.get("parent_device") for sp in sample_positions})
+        parent_devices = list(set(sp.get("parent_device") for sp in sample_positions))
         if len(parent_devices) == 0:
             raise ValueError(f"No sample position(s) beginning with: {position}")
         elif len(parent_devices) > 1:
@@ -231,17 +237,19 @@ def get_sample_position_parent_device(self, position: str) -> Optional[str]:
         return parent_devices[0]
 
     def is_unoccupied_position(self, position: str) -> bool:
-        """Tell if a sample position is unoccupied or not."""
+        """
+        Tell if a sample position is unoccupied or not
+        """
         return (
-            self.get_sample_position_status(position)[0]
-            is not SamplePositionStatus.OCCUPIED
+            not self.get_sample_position_status(position)[0]
+            is SamplePositionStatus.OCCUPIED
         )
 
     def get_available_sample_position(
         self, task_id: ObjectId, position_prefix: str
     ) -> List[Dict[str, Union[str, bool]]]:
         """
-        Check if the position is occupied.
+        Check if the position is occupied
 
         The structure of returned list is ``{"name": str, "need_release": bool}``.
         The entry need_release indicates whether a sample position needs to be released
@@ -286,7 +294,9 @@ def get_available_sample_position(
         return available_sp_names
 
     def lock_sample_position(self, task_id: ObjectId, position: str):
-        """Lock a sample position."""
+        """
+        Lock a sample position
+        """
         sample_status, current_task_id = self.get_sample_position_status(position)
 
         if current_task_id != task_id:
@@ -307,7 +317,9 @@ def lock_sample_position(self, task_id: ObjectId, position: str):
         )
 
     def release_sample_position(self, position: str):
-        """Unlock a sample position."""
+        """
+        Unlock a sample position
+        """
         if self.get_sample_position(position) is None:
             raise ValueError(f"Invalid sample position: {position}")
 
@@ -321,7 +333,9 @@ def release_sample_position(self, position: str):
         )
 
     def get_sample_positions_by_task(self, task_id: Optional[ObjectId]) -> List[str]:
-        """Get the list of sample positions that is locked by a task (given task id)."""
+        """
+        Get the list of sample positions that is locked by a task (given task id)
+        """
         return [
             sample_position["name"]
             for sample_position in self._sample_positions_collection.find(
@@ -342,7 +356,7 @@ def create_sample(
         metadata: Optional[Dict[str, Any]] = None,
     ) -> ObjectId:
         """
-        Create a sample and return its uid in the database.
+        Create a sample and return its uid in the database
 
         Samples with the same name can exist in the database
         """
@@ -376,22 +390,20 @@ def create_sample(
         return cast(ObjectId, result.inserted_id)
 
     def get_sample(self, sample_id: ObjectId) -> Sample:
-        """Get a sample by its id.
+        """Get a sample by its id
 
         Args:
             sample_id (ObjectId): id of the sample within sample collection
 
-        Raises
-        ------
+        Raises:
             ValueError: no sample found with given id
 
-        Returns
-        -------
+        Returns:
             Sample: Sample object for given id
         """
         result = self._sample_collection.find_one({"_id": sample_id})
         if result is None:
-            raise ValueError(f"No sample found with id: {sample_id}")
+            raise ValueError("No sample found with id: {}".format(sample_id))
 
         return Sample(
             sample_id=result["_id"],
@@ -403,7 +415,9 @@ def get_sample(self, sample_id: ObjectId) -> Sample:
         )
 
     def update_sample_task_id(self, sample_id: ObjectId, task_id: Optional[ObjectId]):
-        """Update the task id for a sample."""
+        """
+        Update the task id for a sample
+        """
         result = self._sample_collection.find_one({"_id": sample_id})
         if result is None:
             raise ValueError(f"Cannot find sample with id: {sample_id}")
@@ -419,7 +433,9 @@ def update_sample_task_id(self, sample_id: ObjectId, task_id: Optional[ObjectId]
         )
 
     def move_sample(self, sample_id: ObjectId, position: Optional[str]):
-        """Update the sample with new position."""
+        """
+        Update the sample with new position
+        """
         result = self._sample_collection.find_one({"_id": sample_id})
         if result is None:
             raise ValueError(f"Cannot find sample with id: {sample_id}")
@@ -443,13 +459,12 @@ def move_sample(self, sample_id: ObjectId, position: Optional[str]):
         )
 
     def exists(self, sample_id: Union[ObjectId, str]) -> bool:
-        """Check if a sample exists in the database.
+        """Check if a sample exists in the database
 
         Args:
             sample_id (ObjectId): id of the sample within sample collection
 
-        Returns
-        -------
+        Returns:
             bool: True if sample exists in the database
         """
         return self._sample_collection.count_documents({"_id": ObjectId(sample_id)}) > 0