refactor(robot-server): Type-check robot_server/service/* (#17810)

## Overview

Various type-checking fixes pulled out of EXEC-1342.

## Test Plan and Hands on Testing

* [x] `PATCH /settings/pipettes/:id` still works
* [x] `POST /robot/move` still works, particularly with `"target":
"mount"`, and raises an error if z<30.
* [x] `POST /robot/home` still works, particularly with `"target":
"pipette"`, and raises an error if you do not specify `"mount"` in that
case.
* [x] `POST /settings/log_level/local` still works.
* [x] `GET /wifi/keys` and `POST /wifi/keys` still work.

## Changelog

Mostly adding obvious return type annotations. But there are some
behavioral changes in here, around Pydantic validators and FastAPI
endpoint functions, because it seemed just as easy to fix them as to
annotate them with `#type: ignore`s.

See the individual commit messages in the PR for details.

## Risk assessment

Medium. We need to be careful about the Pydantic validators and FastAPI
endpoint functions.

Author: SyntaxColoring

robot-server/.flake8

@@ -25,16 +25,13 @@ noqa-require-code = true
 # string lints in these modules; remove entries as they are fixed
 per-file-ignores =
     setup.py:ANN,D
-    robot_server/dependencies.py:ANN,D
-    robot_server/util.py:ANN,D
     robot_server/settings.py:ANN,D
     robot_server/robot/*:D
-    robot_server/service/*:ANN,D
+    robot_server/service/*:D
     tests/integration/*:ANN,D
     tests/robot/*:ANN,D
     tests/service/*:ANN,D
     tests/conftest.py:ANN,D
-    tests/test_util.py:ANN,D
 
 # Ignore Python Protocol API files that are used as opaque test input.
 extend-exclude =

robot-server/mypy.ini

@@ -29,37 +29,9 @@ warn_untyped_fields = True
 # TODO(mc, 2021-09-08): fix and remove any / all of the
 # overrides below whenever able
 
-[mypy-robot_server.service.labware.*]
-disallow_untyped_defs = False
-disallow_incomplete_defs = False
-
-[mypy-robot_server.service.legacy.*]
-disallow_untyped_defs = False
-disallow_untyped_calls = False
-disallow_incomplete_defs = False
-
-[mypy-robot_server.service.notifications.*]
-disallow_untyped_defs = False
-disallow_incomplete_defs = False
-
-[mypy-robot_server.service.pipette_offset.*]
-disallow_untyped_defs = False
-disallow_incomplete_defs = False
-
 [mypy-robot_server.service.session.*]
-disallow_untyped_defs = False
-disallow_untyped_calls = False
-disallow_incomplete_defs = False
 no_implicit_reexport = False
 
-[mypy-robot_server.service.tip_length.*]
-disallow_untyped_defs = False
-disallow_incomplete_defs = False
-
-[mypy-robot_server.service.errors]
-disallow_untyped_defs = False
-disallow_incomplete_defs = False
-
 [mypy-tests.service.json_api.*]
 disallow_any_generics = False
 disallow_untyped_defs = False

robot-server/robot_server/service/errors.py

@@ -29,7 +29,7 @@ class ErrorDef(ErrorCreateDef, Enum):
     """An enumeration of ErrorCreateDef Error definitions for use by
     RobotServerError"""
 
-    def __init__(self, e) -> None:
+    def __init__(self, e: Any) -> None:
         super().__init__(**(asdict(e)))
 
 
@@ -48,9 +48,9 @@ def __init__(
         source: Optional[ErrorSource] = None,
         meta: Optional[Dict[str, Any]] = None,
         wrapping: Optional[Sequence[BaseException]] = None,
-        *fmt_args,
-        **fmt_kw_args
-    ):
+        *fmt_args: object,
+        **fmt_kw_args: object
+    ) -> None:
         """
         Constructor.
 

robot-server/robot_server/service/legacy/models/control.py

@@ -1,6 +1,7 @@
 import typing
 from functools import partial
 from enum import Enum
+from typing_extensions import Self
 
 from opentrons import types
 from pydantic import model_validator, ConfigDict, BaseModel, Field
@@ -95,19 +96,20 @@ class RobotMoveTarget(BaseModel):
         "if target is pipette",
     )
 
-    @model_validator(mode="before")
-    @classmethod
-    def root_validator(cls, values):
-        points = values.get("point", [])
-        target = values.get("target")
-        if target == MotionTarget.mount and len(points) == 3 and points[2] < 30:
+    @model_validator(mode="after")
+    def root_validator(self) -> Self:
+        if (
+            self.target == MotionTarget.mount
+            and len(self.point) == 3
+            and self.point[2] < 30
+        ):
             raise ValueError(
                 "Sending a mount to a z position lower than 30 "
                 "can cause a collision with the deck or reach the"
                 " end of the Z axis  movement screw. Z values for"
                 " mount movement must be >= 30"
             )
-        return values
+        return self
 
     model_config = ConfigDict(
         json_schema_extra={
@@ -143,13 +145,12 @@ class RobotHomeTarget(BaseModel):
         " in that case)",
     )
 
-    @model_validator(mode="before")
-    @classmethod
-    def root_validate(cls, values):
+    @model_validator(mode="after")
+    def root_validate(self) -> Self:
         # Make sure that mount is present if target is pipette
-        if values.get("target") == HomeTarget.pipette.value and not values.get("mount"):
+        if self.target == HomeTarget.pipette.value and not self.mount:
             raise ValueError("mount must be specified if target is pipette")
-        return values
+        return self
 
     model_config = ConfigDict(
         json_schema_extra={

robot-server/robot_server/service/legacy/models/motors.py

@@ -51,5 +51,7 @@ class Axes(BaseModel):
 
     @field_validator("axes", mode="before")
     @classmethod
-    def lower_case_motor_name(cls, v):
-        return [m.lower() for m in v]
+    def lower_case_motor_name(cls, v: object) -> object:
+        # type ignore to preserve prior behavior, but I think this is actually
+        # type-unsafe -- I think we need to account for v not being a list[str].
+        return [m.lower() for m in v]  # type: ignore[attr-defined]

robot-server/robot_server/service/legacy/models/networking.py

@@ -172,7 +172,9 @@ class WifiConfiguration(BaseModel):
 
     @field_validator("eapConfig")
     @classmethod
-    def eap_config_validate(cls, v):
+    def eap_config_validate(
+        cls, v: typing.Optional[typing.Dict[str, str]]
+    ) -> typing.Optional[typing.Dict[str, str]]:
         """Custom validator for the eapConfig field"""
         if v is not None:
             if not v.get("eapType"):
@@ -186,7 +188,11 @@ def eap_config_validate(cls, v):
 
     @model_validator(mode="before")
     @classmethod
-    def validate_configuration(cls, values):
+    def validate_configuration(
+        cls,
+        # todo(mm, 2025-03-18): I think values can actually be other types, like str or bool.
+        values: typing.Dict[str, object],
+    ) -> typing.Dict[str, object]:
         """Validate the configuration"""
         security_type = values.get("securityType")
         psk = values.get("psk")

robot-server/robot_server/service/legacy/models/settings.py

@@ -76,7 +76,7 @@ class LogLevels(str, Enum):
 
     """Valid log levels"""
 
-    def __new__(cls, value, level):
+    def __new__(cls, value: str, level: int) -> "LogLevels":
         # https://docs.python.org/3/library/enum.html#when-to-use-new-vs-init
         obj = str.__new__(cls, value)
         obj._value_ = value
@@ -89,7 +89,7 @@ def __new__(cls, value, level):
     error = ("error", logging.ERROR)
 
     @property
-    def level_id(self):
+    def level_id(self) -> int:
         """The log level id as defined in logging lib"""
         return self._level_id
 
@@ -101,8 +101,17 @@ class LogLevel(BaseModel):
 
     @field_validator("log_level", mode="before")
     @classmethod
-    def lower_case_log_keys(cls, value):
-        return value if value is None else LogLevels(value.lower(), None)
+    def lower_case_log_keys(cls, value: object) -> object:
+        if value is None:
+            return value
+        else:
+            # This `type: ignore[call-arg]` is because mypy thinks there needs to be
+            # a second arg here for the int level, but there does not.
+            return LogLevels(  # type: ignore[call-arg]
+                # todo(mm, 2025-03-18): We probably do actually need to check that
+                # the value is a str before calling .lower() on it.
+                value.lower(),  # type: ignore[attr-defined]
+            )
 
 
 class FactoryResetOption(BaseModel):
@@ -209,21 +218,24 @@ class PipetteSettingsUpdate(BaseModel):
 
     @field_validator("setting_fields")
     @classmethod
-    def validate_fields(cls, v):
+    def validate_fields(
+        cls, v: Optional[Dict[str, Optional[PipetteUpdateField]]]
+    ) -> Optional[Dict[str, Optional[PipetteUpdateField]]]:
         """A validator to ensure that values for mutable configs are
         floats and booleans for quirks."""
-        for key, value in v.items():
-            if value is None:
-                pass
-            elif key in model_constants.MUTABLE_CONFIGS_V1:
-                if value.value is not None:
-                    # Must be a float for overriding a config field
-                    value.value = float(value.value)
-            elif key in model_constants.VALID_QUIRKS:
-                if not isinstance(value.value, bool):
-                    raise ValueError(
-                        f"{key} quirk value must " f"be a boolean. Got {value.value}"
-                    )
-            else:
-                raise ValueError(f"{key} is not a valid field or quirk name")
+        if v is not None:
+            for key, value in v.items():
+                if value is None:
+                    pass
+                elif key in model_constants.MUTABLE_CONFIGS_V1:
+                    if value.value is not None:
+                        # Must be a float for overriding a config field
+                        value.value = float(value.value)
+                elif key in model_constants.VALID_QUIRKS:
+                    if not isinstance(value.value, bool):
+                        raise ValueError(
+                            f"{key} quirk value must be a boolean. Got {value.value}"
+                        )
+                else:
+                    raise ValueError(f"{key} is not a valid field or quirk name")
         return v

robot-server/robot_server/service/legacy/routers/control.py

@@ -170,7 +170,7 @@ async def post_robot_light_state(
 
 async def _do_move(
     hardware: HardwareControlAPI, robot_move_target: control.RobotMoveTarget
-):
+) -> Point:
     """Perform the move"""
 
     await hardware.cache_instruments()

robot-server/robot_server/service/legacy/routers/networking.py

@@ -6,7 +6,7 @@
 
 from starlette import status
 from starlette.responses import JSONResponse
-from fastapi import APIRouter, HTTPException, File, Path, UploadFile, Query
+from fastapi import APIRouter, HTTPException, File, Path, UploadFile, Query, Response
 
 from opentrons_shared_data.errors import ErrorCodes
 from opentrons.system import nmcli, wifi
@@ -151,7 +151,7 @@ async def post_wifi_configure(
     response_model=WifiKeyFiles,
     response_model_by_alias=True,
 )
-async def get_wifi_keys():
+async def get_wifi_keys() -> WifiKeyFiles:
     keys = [
         WifiKeyFile(
             uri=f"/wifi/keys/{key.directory}",
@@ -160,12 +160,7 @@ async def get_wifi_keys():
         )
         for key in wifi.list_keys()
     ]
-    # Why not create a WifiKeyFiles? Because validation fails when there's a
-    # pydantic model with attribute named keys. Deep in the guts of pydantic
-    # there's a call to `dict(model)` which raises an exception because `keys`
-    # is not callable, like the `keys` member of dict.
-    # A problem for another time.
-    return {"keys": keys}
+    return WifiKeyFiles(keys=keys)
 
 
 @router.post(
@@ -180,7 +175,10 @@ async def get_wifi_keys():
     status_code=status.HTTP_201_CREATED,
     response_model_exclude_unset=True,
 )
-async def post_wifi_key(key: UploadFile = File(...)):
+async def post_wifi_key(
+    response: Response,
+    key: UploadFile = File(...),
+) -> AddWifiKeyFileResponse:
     key_name = key.filename
     if not key_name:
         raise LegacyErrorResponse(
@@ -189,17 +187,18 @@ async def post_wifi_key(key: UploadFile = File(...)):
 
     add_key_result = wifi.add_key(key_name, key.file.read())
 
-    response = AddWifiKeyFileResponse(
+    response_body = AddWifiKeyFileResponse(
         uri=f"/wifi/keys/{add_key_result.key.directory}",
         id=add_key_result.key.directory,
         name=os.path.basename(add_key_result.key.file),
     )
     if add_key_result.created:
-        return response
+        response.status_code = status.HTTP_201_CREATED
+        return response_body
     else:
-        # We return a JSONResponse because we want the 200 status code.
-        response.message = "Key file already present"
-        return JSONResponse(content=response.model_dump())
+        response.status_code = status.HTTP_200_OK
+        response_body.message = "Key file already present"
+        return response_body
 
 
 @router.delete(
@@ -269,7 +268,7 @@ async def get_eap_options() -> EapOptions:
     responses={status.HTTP_200_OK: {"model": V1BasicResponse}},
     status_code=status.HTTP_207_MULTI_STATUS,
 )
-async def post_wifi_disconnect(wifi_ssid: WifiNetwork):
+async def post_wifi_disconnect(wifi_ssid: WifiNetwork) -> JSONResponse:
     ok, message = await nmcli.wifi_disconnect(wifi_ssid.ssid)
 
     result = V1BasicResponse(message=message)

robot-server/robot_server/service/legacy/routers/pipettes.py

@@ -59,7 +59,9 @@ async def get_pipettes(
 
     attached = hardware.attached_instruments
 
-    def make_pipette(mount: Mount, pipette_dict: PipetteDict, is_ot2: bool):
+    def make_pipette(
+        mount: Mount, pipette_dict: PipetteDict, is_ot2: bool
+    ) -> pipettes.AttachedPipette:
         if is_ot2:
             mount_axis = ot2_axis_to_string(Axis.by_mount(mount))
             plunger_axis = ot2_axis_to_string(Axis.of_plunger(mount))

robot-server/robot_server/service/notifications/publisher_notifier.py

@@ -18,14 +18,16 @@
 class PublisherNotifier:
     """An interface that invokes notification callbacks whenever a generic notify event occurs."""
 
-    def __init__(self, change_notifier: Union[ChangeNotifier, ChangeNotifier_ts]):
+    def __init__(
+        self, change_notifier: Union[ChangeNotifier, ChangeNotifier_ts]
+    ) -> None:
         self._change_notifier = change_notifier
         self._notifier: Optional[asyncio.Task[None]] = None
         self._callbacks: List[Callable[[], Awaitable[None]]] = []
 
     def register_publish_callbacks(
         self, callbacks: List[Callable[[], Awaitable[None]]]
-    ):
+    ) -> None:
         """Extend the list of callbacks with a given list of callbacks."""
         self._callbacks.extend(callbacks)
 

robot-server/robot_server/service/pipette_offset/router.py

@@ -82,7 +82,7 @@ async def delete_specific_pipette_offset_calibration(
     pipette_id: str,
     mount: pip_models.MountType,
     _: Annotated[API, Depends(get_ot2_hardware)],
-):
+) -> None:
     try:
         pipette_offset.delete_pipette_offset_file(
             pipette_id, ot_types.Mount[mount.upper()]

robot-server/robot_server/service/session/command_execution/__init__.py

@@ -1,4 +1,3 @@
-from .base_command_queue import CommandQueue  # noqa: F401
 from .base_executor import CommandExecutor  # noqa: F401
 from .command import (  # noqa: F401
     Command,