refactor(robot-server): Use new Annotated syntax for FastAPI depend…

…encies (#15838)

robot-server/robot_server/client_data/router.py

@@ -89,7 +89,7 @@ async def put_client_data(  # noqa: D103
 )
 async def get_client_data(  # noqa: D103
     key: Key,
-    store: ClientDataStore = fastapi.Depends(get_client_data_store),
+    store: Annotated[ClientDataStore, fastapi.Depends(get_client_data_store)],
 ) -> SimpleBody[ClientData]:
     try:
         return SimpleBody.construct(data=store.get(key))

robot-server/robot_server/client_data/store.py

@@ -1,5 +1,7 @@
 """An in-memory store for arbitrary client-defined JSON objects."""
 
+from typing import Annotated
+
 import fastapi
 
 from server_utils.fastapi_utils.app_state import (
@@ -49,7 +51,7 @@ def delete_all(self) -> None:
 
 
 async def get_client_data_store(
-    app_state: AppState = fastapi.Depends(get_app_state),
+    app_state: Annotated[AppState, fastapi.Depends(get_app_state)],
 ) -> ClientDataStore:
     """A FastAPI dependency to return the server's singleton `ClientDataStore`."""
     store = _app_state_accessor.get_from(app_state)

robot-server/robot_server/commands/get_default_orchestrator.py

@@ -1,5 +1,5 @@
 """Get the default protocol engine."""
-from typing_extensions import Literal
+from typing import Annotated, Literal
 
 from fastapi import Depends, status
 
@@ -34,9 +34,11 @@ class RunActive(ErrorDetails):
 
 
 async def get_default_orchestrator(
-    run_orchestrator_store: RunOrchestratorStore = Depends(get_run_orchestrator_store),
-    hardware_api: HardwareControlAPI = Depends(get_hardware),
-    module_identifier: ModuleIdentifier = Depends(ModuleIdentifier),
+    run_orchestrator_store: Annotated[
+        RunOrchestratorStore, Depends(get_run_orchestrator_store)
+    ],
+    hardware_api: Annotated[HardwareControlAPI, Depends(get_hardware)],
+    module_identifier: Annotated[ModuleIdentifier, Depends(ModuleIdentifier)],
 ) -> RunOrchestrator:
     """Get the default run orchestrator with attached modules loaded."""
     try:

robot-server/robot_server/commands/router.py

@@ -1,6 +1,5 @@
 """Router for top-level /commands endpoints."""
-from typing import List, Optional, cast
-from typing_extensions import Final, Literal
+from typing import Annotated, Final, List, Literal, Optional, cast
 
 from fastapi import APIRouter, Depends, Query, status
 
@@ -65,35 +64,39 @@ class CommandNotFound(ErrorDetails):
 )
 async def create_command(
     request_body: RequestModelWithStatelessCommandCreate,
-    waitUntilComplete: bool = Query(
-        False,
-        description=(
-            "If `false`, return immediately, while the new command is still queued."
-            " If `true`, only return once the new command succeeds or fails,"
-            " or when the timeout is reached. See the `timeout` query parameter."
+    orchestrator: Annotated[RunOrchestrator, Depends(get_default_orchestrator)],
+    waitUntilComplete: Annotated[
+        bool,
+        Query(
+            description=(
+                "If `false`, return immediately, while the new command is still queued."
+                " If `true`, only return once the new command succeeds or fails,"
+                " or when the timeout is reached. See the `timeout` query parameter."
+            ),
         ),
-    ),
-    timeout: Optional[int] = Query(
-        default=None,
-        gt=0,
-        description=(
-            "If `waitUntilComplete` is `true`,"
-            " the maximum time in milliseconds to wait before returning."
-            " The default is infinite."
-            "\n\n"
-            "The timer starts as soon as you enqueue the new command with this request,"
-            " *not* when the new command starts running. So if there are other commands"
-            " in the queue before the new one, they will also count towards the"
-            " timeout."
-            "\n\n"
-            "If the timeout elapses before the command succeeds or fails,"
-            " the command will be returned with its current status."
-            "\n\n"
-            "Compatibility note: on robot software v6.2.0 and older,"
-            " the default was 30 seconds, not infinite."
+    ] = False,
+    timeout: Annotated[
+        Optional[int],
+        Query(
+            gt=0,
+            description=(
+                "If `waitUntilComplete` is `true`,"
+                " the maximum time in milliseconds to wait before returning."
+                " The default is infinite."
+                "\n\n"
+                "The timer starts as soon as you enqueue the new command with this request,"
+                " *not* when the new command starts running. So if there are other commands"
+                " in the queue before the new one, they will also count towards the"
+                " timeout."
+                "\n\n"
+                "If the timeout elapses before the command succeeds or fails,"
+                " the command will be returned with its current status."
+                "\n\n"
+                "Compatibility note: on robot software v6.2.0 and older,"
+                " the default was 30 seconds, not infinite."
+            ),
         ),
-    ),
-    orchestrator: RunOrchestrator = Depends(get_default_orchestrator),
+    ] = None,
 ) -> PydanticResponse[SimpleBody[StatelessCommand]]:
     """Enqueue and execute a command.
 
@@ -133,19 +136,23 @@ async def create_command(
     },
 )
 async def get_commands_list(
-    orchestrator: RunOrchestrator = Depends(get_default_orchestrator),
-    cursor: Optional[int] = Query(
-        None,
-        description=(
-            "The starting index of the desired first command in the list."
-            " If unspecified, a cursor will be selected automatically"
-            " based on the currently running or most recently executed command."
+    orchestrator: Annotated[RunOrchestrator, Depends(get_default_orchestrator)],
+    cursor: Annotated[
+        Optional[int],
+        Query(
+            description=(
+                "The starting index of the desired first command in the list."
+                " If unspecified, a cursor will be selected automatically"
+                " based on the currently running or most recently executed command."
+            ),
         ),
-    ),
-    pageLength: int = Query(
-        _DEFAULT_COMMAND_LIST_LENGTH,
-        description="The maximum number of commands in the list to return.",
-    ),
+    ] = None,
+    pageLength: Annotated[
+        int,
+        Query(
+            description="The maximum number of commands in the list to return.",
+        ),
+    ] = _DEFAULT_COMMAND_LIST_LENGTH,
 ) -> PydanticResponse[SimpleMultiBody[StatelessCommand]]:
     """Get a list of stateless commands.
 
@@ -180,7 +187,7 @@ async def get_commands_list(
 )
 async def get_command(
     commandId: str,
-    orchestrator: RunOrchestrator = Depends(get_default_orchestrator),
+    orchestrator: Annotated[RunOrchestrator, Depends(get_default_orchestrator)],
 ) -> PydanticResponse[SimpleBody[StatelessCommand]]:
     """Get a single stateless command.
 

robot-server/robot_server/data_files/dependencies.py

@@ -1,7 +1,7 @@
 """FastAPI dependencies for data files endpoints."""
 from pathlib import Path
 from asyncio import Lock as AsyncLock
-from typing import Final
+from typing import Annotated, Final
 from anyio import Path as AsyncPath
 
 from fastapi import Depends
@@ -32,8 +32,8 @@
 
 
 async def get_data_files_directory(
-    app_state: AppState = Depends(get_app_state),
-    persistent_directory: Path = Depends(get_active_persistence_directory),
+    app_state: Annotated[AppState, Depends(get_app_state)],
+    persistent_directory: Annotated[Path, Depends(get_active_persistence_directory)],
 ) -> Path:
     """Get the directory to save the protocol files, creating it if needed."""
     async with _data_files_directory_init_lock:
@@ -47,9 +47,9 @@ async def get_data_files_directory(
 
 
 async def get_data_files_store(
-    app_state: AppState = Depends(get_app_state),
-    sql_engine: SQLEngine = Depends(get_sql_engine),
-    data_files_directory: Path = Depends(get_data_files_directory),
+    app_state: Annotated[AppState, Depends(get_app_state)],
+    sql_engine: Annotated[SQLEngine, Depends(get_sql_engine)],
+    data_files_directory: Annotated[Path, Depends(get_data_files_directory)],
 ) -> DataFilesStore:
     """Get a singleton DataFilesStore to keep track of uploaded data files."""
     async with _data_files_store_init_lock:
@@ -61,7 +61,7 @@ async def get_data_files_store(
 
 
 def get_data_file_auto_deleter(
-    data_files_store: DataFilesStore = Depends(get_data_files_store),
+    data_files_store: Annotated[DataFilesStore, Depends(get_data_files_store)],
 ) -> DataFileAutoDeleter:
     """Get a `DataFileAutoDeleter` to delete old data files."""
     return DataFileAutoDeleter(

robot-server/robot_server/data_files/router.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 from pathlib import Path
 from textwrap import dedent
-from typing import Optional, Literal, Union
+from typing import Annotated, Optional, Literal, Union
 
 from fastapi import APIRouter, UploadFile, File, Form, Depends, Response, status
 from opentrons.protocol_reader import FileHasher, FileReaderWriter
@@ -82,19 +82,25 @@ class UnexpectedFileFormat(ErrorDetails):
     },
 )
 async def upload_data_file(
-    file: Optional[UploadFile] = File(default=None, description="Data file to upload"),
-    file_path: Optional[str] = Form(
-        default=None,
-        description="Absolute path to a file on the robot.",
-        alias="filePath",
-    ),
-    data_files_directory: Path = Depends(get_data_files_directory),
-    data_files_store: DataFilesStore = Depends(get_data_files_store),
-    data_file_auto_deleter: DataFileAutoDeleter = Depends(get_data_file_auto_deleter),
-    file_reader_writer: FileReaderWriter = Depends(get_file_reader_writer),
-    file_hasher: FileHasher = Depends(get_file_hasher),
-    file_id: str = Depends(get_unique_id, use_cache=False),
-    created_at: datetime = Depends(get_current_time),
+    data_files_directory: Annotated[Path, Depends(get_data_files_directory)],
+    data_files_store: Annotated[DataFilesStore, Depends(get_data_files_store)],
+    data_file_auto_deleter: Annotated[
+        DataFileAutoDeleter, Depends(get_data_file_auto_deleter)
+    ],
+    file_reader_writer: Annotated[FileReaderWriter, Depends(get_file_reader_writer)],
+    file_hasher: Annotated[FileHasher, Depends(get_file_hasher)],
+    file_id: Annotated[str, Depends(get_unique_id, use_cache=False)],
+    created_at: Annotated[datetime, Depends(get_current_time)],
+    file: Annotated[
+        Optional[UploadFile], File(description="Data file to upload")
+    ] = None,
+    file_path: Annotated[
+        Optional[str],
+        Form(
+            description="Absolute path to a file on the robot.",
+            alias="filePath",
+        ),
+    ] = None,
 ) -> PydanticResponse[SimpleBody[DataFile]]:
     """Save the uploaded data file to persistent storage and update database."""
     if all([file, file_path]):
@@ -162,7 +168,7 @@ async def upload_data_file(
 )
 async def get_data_file_info_by_id(
     dataFileId: str,
-    data_files_store: DataFilesStore = Depends(get_data_files_store),
+    data_files_store: Annotated[DataFilesStore, Depends(get_data_files_store)],
 ) -> PydanticResponse[SimpleBody[DataFile]]:
     """Get data file info by ID.
 
@@ -198,9 +204,9 @@ async def get_data_file_info_by_id(
 )
 async def get_data_file(
     dataFileId: str,
-    data_files_directory: Path = Depends(get_data_files_directory),
-    data_files_store: DataFilesStore = Depends(get_data_files_store),
-    file_reader_writer: FileReaderWriter = Depends(get_file_reader_writer),
+    data_files_directory: Annotated[Path, Depends(get_data_files_directory)],
+    data_files_store: Annotated[DataFilesStore, Depends(get_data_files_store)],
+    file_reader_writer: Annotated[FileReaderWriter, Depends(get_file_reader_writer)],
 ) -> Response:
     """Get the requested data file by id."""
     try:
@@ -228,7 +234,7 @@ async def get_data_file(
     responses={status.HTTP_200_OK: {"model": SimpleMultiBody[str]}},
 )
 async def get_all_data_files(
-    data_files_store: DataFilesStore = Depends(get_data_files_store),
+    data_files_store: Annotated[DataFilesStore, Depends(get_data_files_store)],
 ) -> PydanticResponse[SimpleMultiBody[DataFile]]:
     """Get a list of all data files stored on the robot server.
 

robot-server/robot_server/deck_configuration/fastapi_dependencies.py

@@ -2,7 +2,7 @@
 
 
 from pathlib import Path
-from typing import Optional
+from typing import Annotated, Optional
 
 import fastapi
 
@@ -33,12 +33,14 @@
 
 
 async def get_deck_configuration_store(
-    app_state: AppState = fastapi.Depends(get_app_state),
-    deck_type: DeckType = fastapi.Depends(get_deck_type),
-    persistence_directory: Path = fastapi.Depends(get_active_persistence_directory),
-    deck_configuration_publisher: DeckConfigurationPublisher = fastapi.Depends(
-        get_deck_configuration_publisher
-    ),
+    app_state: Annotated[AppState, fastapi.Depends(get_app_state)],
+    deck_type: Annotated[DeckType, fastapi.Depends(get_deck_type)],
+    persistence_directory: Annotated[
+        Path, fastapi.Depends(get_active_persistence_directory)
+    ],
+    deck_configuration_publisher: Annotated[
+        DeckConfigurationPublisher, fastapi.Depends(get_deck_configuration_publisher)
+    ],
 ) -> DeckConfigurationStore:
     """Return the server's singleton `DeckConfigurationStore`."""
     deck_configuration_store = _accessor.get_from(app_state)
@@ -57,14 +59,14 @@ async def get_deck_configuration_store(
 
 # TODO(mm, 2024-02-07): Resolve the duplication between these two implementations.
 async def get_deck_configuration_store_failsafe(
-    app_state: AppState = fastapi.Depends(get_app_state),
-    deck_type: DeckType = fastapi.Depends(get_deck_type),
-    persistence_directory: Optional[Path] = fastapi.Depends(
-        get_active_persistence_directory_failsafe
-    ),
-    deck_configuration_publisher: DeckConfigurationPublisher = fastapi.Depends(
-        get_deck_configuration_publisher
-    ),
+    app_state: Annotated[AppState, fastapi.Depends(get_app_state)],
+    deck_type: Annotated[DeckType, fastapi.Depends(get_deck_type)],
+    persistence_directory: Annotated[
+        Optional[Path], fastapi.Depends(get_active_persistence_directory_failsafe)
+    ],
+    deck_configuration_publisher: Annotated[
+        DeckConfigurationPublisher, fastapi.Depends(get_deck_configuration_publisher)
+    ],
 ) -> Optional[DeckConfigurationStore]:
     """Return the server's singleton `DeckConfigurationStore`.
 

robot-server/robot_server/deck_configuration/router.py

@@ -2,7 +2,7 @@
 
 
 from datetime import datetime
-from typing import Union
+from typing import Annotated, Union
 
 import fastapi
 from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
@@ -62,9 +62,11 @@
 )
 async def put_deck_configuration(  # noqa: D103
     request_body: RequestModel[models.DeckConfigurationRequest],
-    store: DeckConfigurationStore = fastapi.Depends(get_deck_configuration_store),
-    now: datetime = fastapi.Depends(get_current_time),
-    deck_definition: DeckDefinitionV5 = fastapi.Depends(get_deck_definition),
+    store: Annotated[
+        DeckConfigurationStore, fastapi.Depends(get_deck_configuration_store)
+    ],
+    now: Annotated[datetime, fastapi.Depends(get_current_time)],
+    deck_definition: Annotated[DeckDefinitionV5, fastapi.Depends(get_deck_definition)],
 ) -> PydanticResponse[
     Union[
         SimpleBody[models.DeckConfigurationResponse],
@@ -104,7 +106,9 @@ async def put_deck_configuration(  # noqa: D103
     },
 )
 async def get_deck_configuration(  # noqa: D103
-    store: DeckConfigurationStore = fastapi.Depends(get_deck_configuration_store),
+    store: Annotated[
+        DeckConfigurationStore, fastapi.Depends(get_deck_configuration_store)
+    ],
 ) -> PydanticResponse[SimpleBody[models.DeckConfigurationResponse]]:
     return await PydanticResponse.create(
         content=SimpleBody.construct(data=await store.get())