refactor: Rename Pydantic v1 methods to their v2 counterparts

api/pytest.ini

@@ -6,8 +6,13 @@ markers =
 addopts = --color=yes --strict-markers
 asyncio_mode = auto
 
-# TODO this should be looked into being removed upon updating the Decoy library. The purpose of this warning is to
-#   catch missing attributes, but it raises for any property referenced in a test which accounts for about ~250 warnings
-#   which aren't serving any useful purpose and obscure other warnings.
 filterwarnings =
+    # TODO this should be looked into being removed upon updating the Decoy library. The purpose of this warning is to
+    #   catch missing attributes, but it raises for any property referenced in a test which accounts for about ~250 warnings
+    #   which aren't serving any useful purpose and obscure other warnings.
     ignore::decoy.warnings.MissingSpecAttributeWarning
+    # Pydantic's shims for its legacy v1 methods (e.g. `BaseModel.construct()`)
+    # are not type-checked properly. Forbid them, so we're forced to use their newer
+    # v2 replacements which are type-checked (e.g. ``BaseModel.model_construct()`)
+    error::pydantic.PydanticDeprecatedSince20
+

api/src/opentrons/calibration_storage/deck_configuration.py

@@ -24,9 +24,9 @@ def serialize_deck_configuration(
     cutout_fixture_placements: List[CutoutFixturePlacement], last_modified: datetime
 ) -> bytes:
     """Serialize a deck configuration for storing on the filesystem."""
-    data = _DeckConfigurationModel.construct(
+    data = _DeckConfigurationModel.model_construct(
         cutoutFixtures=[
-            _CutoutFixturePlacementModel.construct(
+            _CutoutFixturePlacementModel.model_construct(
                 cutoutId=e.cutout_id,
                 cutoutFixtureId=e.cutout_fixture_id,
                 opentronsModuleSerialNumber=e.opentrons_module_serial_number,

api/src/opentrons/calibration_storage/file_operators.py

@@ -103,7 +103,7 @@ def save_to_file(
     directory_path.mkdir(parents=True, exist_ok=True)
     file_path = directory_path / f"{file_name}.json"
     json_data = (
-        data.json()
+        data.model_dump_json()
         if isinstance(data, pydantic.BaseModel)
         else json.dumps(data, cls=encoder)
     )
@@ -112,7 +112,7 @@ def save_to_file(
 
 def serialize_pydantic_model(data: pydantic.BaseModel) -> bytes:
     """Safely serialize data from a Pydantic model into a form suitable for storing on disk."""
-    return data.json(by_alias=True).encode("utf-8")
+    return data.model_dump_json(by_alias=True).encode("utf-8")
 
 
 _ModelT = typing.TypeVar("_ModelT", bound=pydantic.BaseModel)

api/src/opentrons/calibration_storage/ot2/tip_length.py

@@ -31,7 +31,7 @@ def _convert_tip_length_model_to_dict(
     # add encoders when converting to a dict.
     dict_of_tip_lengths = {}
     for key, item in to_dict.items():
-        dict_of_tip_lengths[key] = json.loads(item.json())
+        dict_of_tip_lengths[key] = json.loads(item.model_dump_json())
     return dict_of_tip_lengths
 
 
@@ -176,12 +176,14 @@ def delete_tip_length_calibration(
             io.save_to_file(tip_length_dir, pipette_id, dict_of_tip_lengths)
         else:
             io.delete_file(tip_length_dir / f"{pipette_id}.json")
-    elif tiprack_hash and any(tiprack_hash in v.dict() for v in tip_lengths.values()):
+    elif tiprack_hash and any(
+        tiprack_hash in v.model_dump() for v in tip_lengths.values()
+    ):
         # NOTE this is for backwards compatibilty only
         # TODO delete this check once the tip_length DELETE router
         # no longer depends on a tiprack hash
         for k, v in tip_lengths.items():
-            if tiprack_hash in v.dict():
+            if tiprack_hash in v.model_dump():
                 tip_lengths.pop(k)
         if tip_lengths:
             dict_of_tip_lengths = _convert_tip_length_model_to_dict(tip_lengths)

api/src/opentrons/execute.py

@@ -560,7 +560,9 @@ def _create_live_context_pe(
     # Non-async would use call_soon_threadsafe(), which makes the waiting harder.
     async def add_all_extra_labware() -> None:
         for labware_definition_dict in extra_labware.values():
-            labware_definition = LabwareDefinition.parse_obj(labware_definition_dict)
+            labware_definition = LabwareDefinition.model_validate(
+                labware_definition_dict
+            )
             pe.add_labware_definition(labware_definition)
 
     # Add extra_labware to ProtocolEngine, being careful not to modify ProtocolEngine from this

api/src/opentrons/hardware_control/emulation/module_server/client.py

@@ -66,7 +66,7 @@ async def read(self) -> Message:
         """Read a message from the module server."""
         try:
             b = await self._reader.readuntil(MessageDelimiter)
-            m: Message = Message.parse_raw(b)
+            m: Message = Message.model_validate_json(b)
             return m
         except LimitOverrunError as e:
             raise ModuleServerClientError(str(e))

api/src/opentrons/hardware_control/emulation/module_server/server.py

@@ -53,7 +53,9 @@ def on_server_connected(
         self._connections[identifier] = connection
         for c in self._clients:
             c.write(
-                Message(status="connected", connections=[connection]).json().encode()
+                Message(status="connected", connections=[connection])
+                .model_dump_json()
+                .encode()
             )
             c.write(b"\n")
 
@@ -72,7 +74,7 @@ def on_server_disconnected(self, identifier: str) -> None:
             for c in self._clients:
                 c.write(
                     Message(status="disconnected", connections=[connection])
-                    .json()
+                    .model_dump_json()
                     .encode()
                 )
                 c.write(MessageDelimiter)
@@ -95,7 +97,7 @@ async def _handle_connection(
         # A client connected. Send a dump of all connected modules.
         m = Message(status="dump", connections=list(self._connections.values()))
 
-        writer.write(m.json().encode())
+        writer.write(m.model_dump_json().encode())
         writer.write(MessageDelimiter)
 
         self._clients.add(writer)

api/src/opentrons/hardware_control/instruments/ot2/instrument_calibration.py

@@ -123,7 +123,8 @@ def load_tip_length_for_pipette(
 ) -> TipLengthCalibration:
     if isinstance(tiprack, LabwareDefinition):
         tiprack = typing.cast(
-            "TypeDictLabwareDef", tiprack.dict(exclude_none=True, exclude_unset=True)
+            "TypeDictLabwareDef",
+            tiprack.model_dump(exclude_none=True, exclude_unset=True),
         )
 
     tip_length_data = calibration_storage.load_tip_length_calibration(

api/src/opentrons/hardware_control/instruments/ot2/pipette.py

@@ -96,7 +96,7 @@ def __init__(
         use_old_aspiration_functions: bool = False,
     ) -> None:
         self._config = config
-        self._config_as_dict = config.dict()
+        self._config_as_dict = config.model_dump()
         self._pipette_offset = pipette_offset_cal
         self._pipette_type = self._config.pipette_type
         self._pipette_version = self._config.version
@@ -273,15 +273,15 @@ def update_config_item(
             self._config, elements, liquid_class
         )
         # Update the cached dict representation
-        self._config_as_dict = self._config.dict()
+        self._config_as_dict = self._config.model_dump()
 
     def reload_configurations(self) -> None:
         self._config = load_pipette_data.load_definition(
             self._pipette_model.pipette_type,
             self._pipette_model.pipette_channels,
             self._pipette_model.pipette_version,
         )
-        self._config_as_dict = self._config.dict()
+        self._config_as_dict = self._config.model_dump()
 
     def reset_state(self) -> None:
         self._current_volume = 0.0
@@ -656,8 +656,8 @@ def _reload_and_check_skip(
         # Same config, good enough
         return attached_instr, True
     else:
-        newdict = new_config.dict()
-        olddict = attached_instr.config.dict()
+        newdict = new_config.model_dump()
+        olddict = attached_instr.config.model_dump()
         changed: Set[str] = set()
         for k in newdict.keys():
             if newdict[k] != olddict[k]:

api/src/opentrons/hardware_control/instruments/ot3/gripper.py

@@ -318,8 +318,8 @@ def _reload_gripper(
         # Same config, good enough
         return attached_instr, True
     else:
-        newdict = new_config.dict()
-        olddict = attached_instr.config.dict()
+        newdict = new_config.model_dump()
+        olddict = attached_instr.config.model_dump()
         changed: Set[str] = set()
         for k in newdict.keys():
             if newdict[k] != olddict[k]:

api/src/opentrons/hardware_control/instruments/ot3/pipette.py

@@ -79,7 +79,7 @@ def __init__(
         use_old_aspiration_functions: bool = False,
     ) -> None:
         self._config = config
-        self._config_as_dict = config.dict()
+        self._config_as_dict = config.model_dump()
         self._plunger_motor_current = config.plunger_motor_configurations
         self._pick_up_configurations = config.pick_up_tip_configurations
         self._plunger_homing_configurations = config.plunger_homing_configurations
@@ -251,7 +251,7 @@ def reload_configurations(self) -> None:
             self._pipette_model.pipette_channels,
             self._pipette_model.pipette_version,
         )
-        self._config_as_dict = self._config.dict()
+        self._config_as_dict = self._config.model_dump()
 
     def reset_state(self) -> None:
         self._current_volume = 0.0
@@ -770,8 +770,8 @@ def _reload_and_check_skip(
         # Same config, good enough
         return attached_instr, True
     else:
-        newdict = new_config.dict()
-        olddict = attached_instr.config.dict()
+        newdict = new_config.model_dump()
+        olddict = attached_instr.config.model_dump()
         changed: Set[str] = set()
         for k in newdict.keys():
             if newdict[k] != olddict[k]:

api/src/opentrons/hardware_control/ot3_calibration.py

@@ -968,7 +968,7 @@ def load_attitude_matrix(to_default: bool = True) -> DeckCalibration:
         return DeckCalibration(
             attitude=apply_machine_transform(calibration_data.attitude),
             source=calibration_data.source,
-            status=types.CalibrationStatus(**calibration_data.status.dict()),
+            status=types.CalibrationStatus(**calibration_data.status.model_dump()),
             belt_attitude=calibration_data.attitude,
             last_modified=calibration_data.lastModified,
             pipette_calibrated_with=calibration_data.pipetteCalibratedWith,

api/src/opentrons/hardware_control/robot_calibration.py

@@ -154,7 +154,7 @@ def load_attitude_matrix() -> DeckCalibration:
         return DeckCalibration(
             attitude=calibration_data.attitude,
             source=calibration_data.source,
-            status=types.CalibrationStatus(**calibration_data.status.dict()),
+            status=types.CalibrationStatus(**calibration_data.status.model_dump()),
             last_modified=calibration_data.last_modified,
             pipette_calibrated_with=calibration_data.pipette_calibrated_with,
             tiprack=calibration_data.tiprack,

api/src/opentrons/protocol_api/core/engine/labware.py

@@ -92,12 +92,14 @@ def get_name(self) -> str:
 
     def get_definition(self) -> LabwareDefinitionDict:
         """Get the labware's definition as a plain dictionary."""
-        return cast(LabwareDefinitionDict, self._definition.dict(exclude_none=True))
+        return cast(
+            LabwareDefinitionDict, self._definition.model_dump(exclude_none=True)
+        )
 
     def get_parameters(self) -> LabwareParametersDict:
         return cast(
             LabwareParametersDict,
-            self._definition.parameters.dict(exclude_none=True),
+            self._definition.parameters.model_dump(exclude_none=True),
         )
 
     def get_quirks(self) -> List[str]:
@@ -118,7 +120,7 @@ def set_calibration(self, delta: Point) -> None:
                 details={"kind": "labware-not-in-slot"},
             )
 
-        request = LabwareOffsetCreate.construct(
+        request = LabwareOffsetCreate.model_construct(
             definitionUri=self.get_uri(),
             location=offset_location,
             vector=LabwareOffsetVector(x=delta.x, y=delta.y, z=delta.z),

api/src/opentrons/protocol_api/core/engine/protocol.py

@@ -193,7 +193,7 @@ def add_labware_definition(
     ) -> LabwareLoadParams:
         """Add a labware definition to the set of loadable definitions."""
         uri = self._engine_client.add_labware_definition(
-            LabwareDefinition.parse_obj(definition)
+            LabwareDefinition.model_validate(definition)
         )
         return LabwareLoadParams.from_uri(uri)
 

api/src/opentrons/protocol_engine/commands/absorbance_reader/read.py

@@ -124,7 +124,9 @@ async def execute(  # noqa: C901
                     )
                     asbsorbance_result[wavelength] = converted_values
                     transform_results.append(
-                        ReadData.construct(wavelength=wavelength, data=converted_values)
+                        ReadData.model_construct(
+                            wavelength=wavelength, data=converted_values
+                        )
                     )
         # Handle the virtual module case for data creation (all zeroes)
         elif self._state_view.config.use_virtual_modules:
@@ -138,7 +140,9 @@ async def execute(  # noqa: C901
                     )
                     asbsorbance_result[wavelength] = converted_values
                     transform_results.append(
-                        ReadData.construct(wavelength=wavelength, data=converted_values)
+                        ReadData.model_construct(
+                            wavelength=wavelength, data=converted_values
+                        )
                     )
             else:
                 raise CannotPerformModuleAction(
@@ -153,7 +157,7 @@ async def execute(  # noqa: C901
         file_ids: list[str] = []
         if params.fileName is not None:
             # Create the Plate Reader Transform
-            plate_read_result = PlateReaderData.construct(
+            plate_read_result = PlateReaderData.model_construct(
                 read_results=transform_results,
                 reference_wavelength=abs_reader_substate.reference_wavelength,
                 start_time=start_time,

api/src/opentrons/protocol_engine/commands/calibration/calibrate_gripper.py

@@ -125,8 +125,8 @@ async def execute(
             calibration_data = result
 
         return SuccessData(
-            public=CalibrateGripperResult.construct(
-                jawOffset=Vec3f.construct(
+            public=CalibrateGripperResult.model_construct(
+                jawOffset=Vec3f.model_construct(
                     x=probe_offset.x, y=probe_offset.y, z=probe_offset.z
                 ),
                 savedCalibration=calibration_data,

api/src/opentrons/protocol_engine/commands/calibration/calibrate_pipette.py

@@ -65,8 +65,8 @@ async def execute(
         await ot3_api.save_instrument_offset(mount=ot3_mount, delta=pipette_offset)
 
         return SuccessData(
-            public=CalibratePipetteResult.construct(
-                pipetteOffset=InstrumentOffsetVector.construct(
+            public=CalibratePipetteResult.model_construct(
+                pipetteOffset=InstrumentOffsetVector.model_construct(
                     x=pipette_offset.x, y=pipette_offset.y, z=pipette_offset.z
                 )
             ),

api/src/opentrons/protocol_engine/commands/configure_nozzle_layout.py

@@ -61,9 +61,11 @@ async def execute(
         self, params: ConfigureNozzleLayoutParams
     ) -> SuccessData[ConfigureNozzleLayoutResult]:
         """Check that requested pipette can support the requested nozzle layout."""
-        primary_nozzle = params.configurationParams.dict().get("primaryNozzle")
-        front_right_nozzle = params.configurationParams.dict().get("frontRightNozzle")
-        back_left_nozzle = params.configurationParams.dict().get("backLeftNozzle")
+        primary_nozzle = params.configurationParams.model_dump().get("primaryNozzle")
+        front_right_nozzle = params.configurationParams.model_dump().get(
+            "frontRightNozzle"
+        )
+        back_left_nozzle = params.configurationParams.model_dump().get("backLeftNozzle")
         nozzle_params = await self._tip_handler.available_for_nozzle_layout(
             pipette_id=params.pipetteId,
             style=params.configurationParams.style,

api/src/opentrons/protocol_engine/commands/custom.py

@@ -44,7 +44,7 @@ class CustomImplementation(
     async def execute(self, params: CustomParams) -> SuccessData[CustomResult]:
         """A custom command does nothing when executed directly."""
         return SuccessData(
-            public=CustomResult.construct(),
+            public=CustomResult.model_construct(),
         )
 
 

api/src/opentrons/protocol_engine/commands/movement_common.py

@@ -182,7 +182,7 @@ async def move_to_well(
             state_update=StateUpdate().clear_all_pipette_locations(),
         )
     else:
-        deck_point = DeckPoint.construct(x=position.x, y=position.y, z=position.z)
+        deck_point = DeckPoint.model_construct(x=position.x, y=position.y, z=position.z)
         return SuccessData(
             public=DestinationPositionResult(
                 position=deck_point,
@@ -222,7 +222,7 @@ async def move_relative(
             state_update=StateUpdate().clear_all_pipette_locations(),
         )
     else:
-        deck_point = DeckPoint.construct(x=position.x, y=position.y, z=position.z)
+        deck_point = DeckPoint.model_construct(x=position.x, y=position.y, z=position.z)
         return SuccessData(
             public=DestinationPositionResult(
                 position=deck_point,
@@ -277,7 +277,7 @@ async def move_to_addressable_area(
             .set_addressable_area_used(addressable_area_name=addressable_area_name),
         )
     else:
-        deck_point = DeckPoint.construct(x=x, y=y, z=z)
+        deck_point = DeckPoint.model_construct(x=x, y=y, z=z)
         return SuccessData(
             public=DestinationPositionResult(position=deck_point),
             state_update=StateUpdate()
@@ -324,7 +324,7 @@ async def move_to_coordinates(
             state_update=StateUpdate().clear_all_pipette_locations(),
         )
     else:
-        deck_point = DeckPoint.construct(x=x, y=y, z=z)
+        deck_point = DeckPoint.model_construct(x=x, y=y, z=z)
 
         return SuccessData(
             public=DestinationPositionResult(position=DeckPoint(x=x, y=y, z=z)),

api/src/opentrons/protocol_engine/commands/touch_tip.py

@@ -154,7 +154,7 @@ async def execute(
             waypoints=touch_waypoints,
             speed=touch_speed,
         )
-        final_deck_point = DeckPoint.construct(
+        final_deck_point = DeckPoint.model_construct(
             x=final_point.x, y=final_point.y, z=final_point.z
         )
         state_update = center_result.state_update.set_pipette_location(

api/src/opentrons/protocol_engine/errors/error_occurrence.py

@@ -29,7 +29,7 @@ def from_failed(
             wrappedErrors = [
                 cls.from_failed(id, createdAt, err) for err in error.wrapping
             ]
-        return cls.construct(
+        return cls.model_construct(
             id=id,
             createdAt=createdAt,
             errorType=type(error).__name__,