api(src): update deprecated code

api/src/opentrons/calibration_storage/deck_configuration.py

@@ -23,9 +23,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
             )
             for e in cutout_fixture_placements

api/src/opentrons/cli/analyze.py

@@ -82,18 +82,18 @@ async def _analyze(
     )
     analysis = await runner.run(deck_configuration=[], protocol_source=protocol_source)
     if json_output:
-        results = AnalyzeResults.construct(
+        results = AnalyzeResults.model_construct(
             createdAt=datetime.now(tz=timezone.utc),
             files=[
-                ProtocolFile.construct(name=f.path.name, role=f.role)
+                ProtocolFile.model_construct(name=f.path.name, role=f.role)
                 for f in protocol_source.files
             ],
             config=(
-                JsonConfig.construct(
+                JsonConfig.model_construct(
                     schemaVersion=protocol_source.config.schema_version
                 )
                 if isinstance(protocol_source.config, JsonProtocolConfig)
-                else PythonConfig.construct(
+                else PythonConfig.model_construct(
                     apiVersion=protocol_source.config.api_version
                 )
             ),

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

@@ -34,6 +34,7 @@
     Command,
     CommandParams,
     CommandCreate,
+    CommandCreateAdatper,
     CommandResult,
     CommandType,
     CommandPrivateResult,
@@ -321,6 +322,7 @@
     "Command",
     "CommandParams",
     "CommandCreate",
+    "CommandCreateAdatper",
     "CommandResult",
     "CommandType",
     "CommandPrivateResult",

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

@@ -118,8 +118,8 @@ async def execute(self, params: CalibrateGripperParams) -> CalibrateGripperResul
             )
             calibration_data = result
 
-        return CalibrateGripperResult.construct(
-            jawOffset=Vec3f.construct(
+        return 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(self, params: CalibratePipetteParams) -> CalibratePipetteResul
 
         await ot3_api.save_instrument_offset(mount=ot3_mount, delta=pipette_offset)
 
-        return CalibratePipetteResult.construct(
-            pipetteOffset=InstrumentOffsetVector.construct(
+        return 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/command_unions.py

@@ -3,7 +3,7 @@
 from typing import Union, TypeVar
 from typing_extensions import Annotated
 
-from pydantic import Field
+from pydantic import Field, TypeAdapter
 
 from . import heater_shaker
 from . import magnetic_module
@@ -542,6 +542,11 @@
     Field(discriminator="commandType"),
 ]
 
+# Each time a TypeAdapter is instantiated, it will construct a new validator and
+# serializer. To improve performance, TypeAdapters are instantiated once.
+# See https://docs.pydantic.dev/latest/concepts/performance/#typeadapter-instantiated-once
+CommandCreateAdatper: TypeAdapter[CommandCreate] = TypeAdapter(CommandCreate)  # type: ignore[arg-type]
+
 CommandResult = Union[
     AspirateResult,
     AspirateInPlaceResult,

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

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

api/src/opentrons/protocol_engine/slot_standardization.py

@@ -35,9 +35,9 @@ def standardize_labware_offset(
     original: LabwareOffsetCreate, robot_type: RobotType
 ) -> LabwareOffsetCreate:
     """Convert the deck slot in the given `LabwareOffsetCreate` to match the given robot type."""
-    return original.copy(
+    return original.model_copy(
         update={
-            "location": original.location.copy(
+            "location": original.location.model_copy(
                 update={
                     "slotName": original.location.slotName.to_equivalent_for_robot_type(
                         robot_type
@@ -70,40 +70,40 @@ def standardize_command(
 def _standardize_load_labware(
     original: commands.LoadLabwareCreate, robot_type: RobotType
 ) -> commands.LoadLabwareCreate:
-    params = original.params.copy(
+    params = original.params.model_copy(
         update={
             "location": _standardize_labware_location(
                 original.params.location, robot_type
             )
         }
     )
-    return original.copy(update={"params": params})
+    return original.model_copy(update={"params": params})
 
 
 def _standardize_load_module(
     original: commands.LoadModuleCreate, robot_type: RobotType
 ) -> commands.LoadModuleCreate:
-    params = original.params.copy(
+    params = original.params.model_copy(
         update={
             "location": _standardize_deck_slot_location(
                 original.params.location, robot_type
             )
         }
     )
-    return original.copy(update={"params": params})
+    return original.model_copy(update={"params": params})
 
 
 def _standardize_move_labware(
     original: commands.MoveLabwareCreate, robot_type: RobotType
 ) -> commands.MoveLabwareCreate:
-    params = original.params.copy(
+    params = original.params.model_copy(
         update={
             "newLocation": _standardize_labware_location(
                 original.params.newLocation, robot_type
             )
         }
     )
-    return original.copy(update={"params": params})
+    return original.model_copy(update={"params": params})
 
 
 _standardize_command_functions: Dict[
@@ -135,6 +135,6 @@ def _standardize_labware_location(
 def _standardize_deck_slot_location(
     original: DeckSlotLocation, robot_type: RobotType
 ) -> DeckSlotLocation:
-    return original.copy(
+    return original.model_copy(
         update={"slotName": original.slotName.to_equivalent_for_robot_type(robot_type)}
     )

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

@@ -226,15 +226,15 @@ def handle_action(self, action: Action) -> None:  # noqa: C901
             # request > command mapping, figure out how to type precisely
             # (or wait for a future mypy version that can figure it out).
             # For now, unit tests cover mapping every request type
-            queued_command = action.request._CommandCls.construct(
+            queued_command = action.request._CommandCls(
                 id=action.command_id,
                 key=(
                     action.request.key
                     if action.request.key is not None
                     else (action.request_hash or action.command_id)
                 ),
                 createdAt=action.created_at,
-                params=action.request.params,
+                params=action.request.params,  # type: ignore[arg-type]
                 intent=action.request.intent,
                 status=CommandStatus.QUEUED,
             )
@@ -527,7 +527,7 @@ def get_error(self) -> Optional[ErrorOccurrence]:
         finish_error = self._state.finish_error
 
         if run_error and finish_error:
-            combined_error = ErrorOccurrence.construct(
+            combined_error = ErrorOccurrence(
                 id=finish_error.id,
                 createdAt=finish_error.createdAt,
                 errorType="RunAndFinishFailed",

api/src/opentrons/protocol_engine/state/labware.py

@@ -129,7 +129,7 @@ def __init__(
             for fixed_labware in deck_fixed_labware
         }
         labware_by_id = {
-            fixed_labware.labware_id: LoadedLabware.construct(
+            fixed_labware.labware_id: LoadedLabware.model_construct(
                 id=fixed_labware.labware_id,
                 location=fixed_labware.location,
                 loadName=fixed_labware.definition.parameters.loadName,
@@ -156,7 +156,7 @@ def handle_action(self, action: Action) -> None:
             self._handle_command(action.command)
 
         elif isinstance(action, AddLabwareOffsetAction):
-            labware_offset = LabwareOffset.construct(
+            labware_offset = LabwareOffset.model_construct(
                 id=action.labware_offset_id,
                 createdAt=action.created_at,
                 definitionUri=action.request.definitionUri,
@@ -190,7 +190,7 @@ def _handle_command(self, command: Command) -> None:
 
             self._state.labware_by_id[
                 command.result.labwareId
-            ] = LoadedLabware.construct(
+            ] = LoadedLabware.model_construct(
                 id=command.result.labwareId,
                 location=command.params.location,
                 loadName=command.result.definition.parameters.loadName,

api/src/opentrons/protocol_engine/state/modules.py

@@ -530,7 +530,7 @@ def get(self, module_id: str) -> LoadedModule:
             DeckSlotLocation(slotName=slot_name) if slot_name is not None else None
         )
 
-        return LoadedModule.construct(
+        return LoadedModule.model_construct(
             id=module_id,
             location=location,
             model=attached_module.definition.model,

api/src/opentrons/protocol_engine/state/state.py

@@ -115,7 +115,7 @@ def get_summary(self) -> StateSummary:
         """Get protocol run data."""
         error = self._commands.get_error()
         # TODO maybe add summary here for AA
-        return StateSummary.construct(
+        return StateSummary.model_construct(
             status=self._commands.get_status(),
             errors=[] if error is None else [error],
             pipettes=self._pipettes.get_all(),

api/src/opentrons/protocol_runner/json_translator.py

@@ -35,7 +35,6 @@ class CommandTranslatorError(Exception):
 # serializer. To improve performance, TypeAdapters are instantiated once.
 # See https://docs.pydantic.dev/latest/concepts/performance/#typeadapter-instantiated-once
 LabwareLocationAdapter: TypeAdapter[LabwareLocation] = TypeAdapter(LabwareLocation)  # type: ignore[arg-type]
-CommandCreateAdatper: TypeAdapter[pe_commands.CommandCreate] = TypeAdapter(pe_commands.CommandCreate)  # type: ignore[arg-type]
 
 
 def _translate_labware_command(
@@ -195,7 +194,7 @@ def _translate_simple_command(
         else:
             dict_command["commandType"] = "waitForDuration"
 
-    return CommandCreateAdatper.validate_python(dict_command)
+    return pe_commands.CommandCreateAdatper.validate_python(dict_command)
 
 
 class JsonTranslator: