Remove fastapi-utls

lib/galaxy/dependencies/pinned-requirements.txt

@@ -64,7 +64,6 @@ ecdsa==0.18.0 ; python_version >= "3.8" and python_version < "3.12"
 edam-ontology==1.25.2 ; python_version >= "3.8" and python_version < "3.12"
 email-validator==2.1.0.post1 ; python_version >= "3.8" and python_version < "3.12"
 exceptiongroup==1.2.0 ; python_version >= "3.8" and python_version < "3.11"
-fastapi-utils==0.2.1 ; python_version >= "3.8" and python_version < "3.12"
 fastapi==0.98.0 ; python_version >= "3.8" and python_version < "3.12"
 filelock==3.13.1 ; python_version >= "3.8" and python_version < "3.12"
 frozenlist==1.4.1 ; python_version >= "3.8" and python_version < "3.12"

lib/galaxy/webapps/galaxy/api/__init__.py

@@ -38,7 +38,6 @@
     APIKeyHeader,
     APIKeyQuery,
 )
-from fastapi_utils.cbv import cbv
 from pydantic import ValidationError
 from pydantic.main import BaseModel
 from starlette.datastructures import Headers
@@ -442,15 +441,6 @@ def _handle_galaxy_kwd(self, kwd):
 
         return kwd
 
-    @property
-    def cbv(self):
-        """Short-hand for frequently used Galaxy-pattern of FastAPI class based views.
-
-        Creates a class-based view for for this router, for more information see:
-        https://fastapi-utils.davidmontague.xyz/user-guide/class-based-views/
-        """
-        return cbv(self)
-
 
 class Router(FrameworkRouter):
     admin_user_dependency = AdminUserRequired

lib/galaxy/webapps/galaxy/api/authenticate.py

@@ -50,15 +50,13 @@ def options(self, trans: GalaxyWebTransaction, **kwd):
         # trans.response.headers['Access-Control-Allow-Methods'] = 'POST, PUT, GET, OPTIONS, DELETE'
 
 
-@router.cbv
-class FastAPIAuthenticate:
-    authentication_service: AuthenticationService = depends(AuthenticationService)
-
-    @router.get(
-        "/api/authenticate/baseauth",
-        summary="Returns returns an API key for authenticated user based on BaseAuth headers.",
-    )
-    def get_api_key(self, request: Request) -> APIKeyResponse:
-        authorization = request.headers.get("Authorization")
-        auth = {"HTTP_AUTHORIZATION": authorization}
-        return self.authentication_service.get_api_key(auth, request)
+@router.get(
+    "/api/authenticate/baseauth",
+    summary="Returns returns an API key for authenticated user based on BaseAuth headers.",
+)
+def get_api_key(
+    request: Request, authentication_service: AuthenticationService = depends(AuthenticationService)
+) -> APIKeyResponse:
+    authorization = request.headers.get("Authorization")
+    auth = {"HTTP_AUTHORIZATION": authorization}
+    return authentication_service.get_api_key(auth, request)

lib/galaxy/webapps/galaxy/api/cloud.py

@@ -32,72 +32,69 @@
 router = Router(tags=["cloud"])
 
 
-@router.cbv
-class FastAPICloudController:
-    cloud_manager: CloudManager = depends(CloudManager)
-    datasets_serializer: DatasetSerializer = depends(DatasetSerializer)
+@router.get(
+    "/api/cloud/storage",
+    summary="Lists cloud-based buckets (e.g., S3 bucket, Azure blob) user has defined. Is not yet implemented",
+    deprecated=True,
+)
+def index(
+    response: Response,
+):
+    # TODO: This can be implemented leveraging PluggedMedia objects (part of the user-based object store project)
+    response.status_code = status.HTTP_501_NOT_IMPLEMENTED
+    return StatusCode(detail="Not yet implemented.", status=501)
 
-    @router.get(
-        "/api/cloud/storage",
-        summary="Lists cloud-based buckets (e.g., S3 bucket, Azure blob) user has defined. Is not yet implemented",
-        deprecated=True,
-    )
-    def index(
-        self,
-        response: Response,
-    ):
-        # TODO: This can be implemented leveraging PluggedMedia objects (part of the user-based object store project)
-        response.status_code = status.HTTP_501_NOT_IMPLEMENTED
-        return StatusCode(detail="Not yet implemented.", status=501)
 
-    @router.post(
-        "/api/cloud/storage/get",
-        summary="Gets given objects from a given cloud-based bucket to a Galaxy history.",
-        deprecated=True,
+@router.post(
+    "/api/cloud/storage/get",
+    summary="Gets given objects from a given cloud-based bucket to a Galaxy history.",
+    deprecated=True,
+)
+def get(
+    payload: CloudObjects = Body(default=Required),
+    trans: ProvidesHistoryContext = DependsOnTrans,
+    cloud_manager: CloudManager = depends(CloudManager),
+    datasets_serializer: DatasetSerializer = depends(DatasetSerializer),
+) -> DatasetSummaryList:
+    datasets = cloud_manager.get(
+        trans=trans,
+        history_id=payload.history_id,
+        bucket_name=payload.bucket,
+        objects=payload.objects,
+        authz_id=payload.authz_id,
+        input_args=payload.input_args,
     )
-    def get(
-        self,
-        payload: CloudObjects = Body(default=Required),
-        trans: ProvidesHistoryContext = DependsOnTrans,
-    ) -> DatasetSummaryList:
-        datasets = self.cloud_manager.get(
-            trans=trans,
-            history_id=payload.history_id,
-            bucket_name=payload.bucket,
-            objects=payload.objects,
-            authz_id=payload.authz_id,
-            input_args=payload.input_args,
-        )
-        rtv = []
-        for dataset in datasets:
-            rtv.append(self.datasets_serializer.serialize_to_view(dataset, view="summary"))
-        return DatasetSummaryList.construct(__root__=rtv)
+    rtv = []
+    for dataset in datasets:
+        rtv.append(datasets_serializer.serialize_to_view(dataset, view="summary"))
+    return DatasetSummaryList.construct(__root__=rtv)
 
-    @router.post(
-        "/api/cloud/storage/send",
-        summary="Sends given dataset(s) in a given history to a given cloud-based bucket.",
-        deprecated=True,
-    )
-    def send(
-        self,
-        payload: CloudDatasets = Body(default=Required),
-        trans: ProvidesHistoryContext = DependsOnTrans,
-    ) -> CloudDatasetsResponse:
-        log.info(
-            msg="Received api/send request for `{}` datasets using authnz with id `{}`, and history `{}`."
-            "".format(
-                "all the dataset in the given history" if not payload.dataset_ids else len(payload.dataset_ids),
-                payload.authz_id,
-                payload.history_id,
-            )
-        )
 
-        sent, failed = self.cloud_manager.send(
-            trans=trans,
-            history_id=payload.history_id,
-            bucket_name=payload.bucket,
-            authz_id=payload.authz_id,
-            dataset_ids=payload.dataset_ids,
-            overwrite_existing=payload.overwrite_existing,
+@router.post(
+    "/api/cloud/storage/send",
+    summary="Sends given dataset(s) in a given history to a given cloud-based bucket.",
+    deprecated=True,
+)
+def send(
+    payload: CloudDatasets = Body(default=Required),
+    trans: ProvidesHistoryContext = DependsOnTrans,
+    cloud_manager: CloudManager = depends(CloudManager),
+) -> CloudDatasetsResponse:
+    log.info(
+        msg="Received api/send request for `{}` datasets using authnz with id `{}`, and history `{}`."
+        "".format(
+            "all the dataset in the given history" if not payload.dataset_ids else len(payload.dataset_ids),
+            payload.authz_id,
+            payload.history_id,
         )
-        return CloudDatasetsResponse(sent_dataset_labels=sent, failed_dataset_labels=failed, bucket_name=payload.bucket)
+    )
+
+    sent, failed = cloud_manager.send(
+        trans=trans,
+        history_id=payload.history_id,
+        bucket_name=payload.bucket,
+        authz_id=payload.authz_id,
+        dataset_ids=payload.dataset_ids,
+        overwrite_existing=payload.overwrite_existing,
+    )
+    return CloudDatasetsResponse(sent_dataset_labels=sent, failed_dataset_labels=failed, bucket_name=payload.bucket)

lib/galaxy/webapps/galaxy/api/configuration.py

@@ -39,84 +39,97 @@
 )
 
 
-@router.cbv
-class FastAPIConfiguration:
-    configuration_manager: ConfigurationManager = depends(ConfigurationManager)
-
-    @router.get(
-        "/api/whoami",
-        summary="Return information about the current authenticated user",
-        response_description="Information about the current authenticated user",
-    )
-    def whoami(self, trans: ProvidesUserContext = DependsOnTrans) -> Optional[UserModel]:
-        """Return information about the current authenticated user."""
-        return _user_to_model(trans.user)
-
-    @router.get(
-        "/api/configuration",
-        summary="Return an object containing exposable configuration settings",
-        response_description="Object containing exposable configuration settings",
-    )
-    def index(
-        self,
-        trans: ProvidesUserContext = DependsOnTrans,
-        view: Optional[str] = SerializationViewQueryParam,
-        keys: Optional[str] = SerializationKeysQueryParam,
-    ) -> Dict[str, Any]:
-        """
-        Return an object containing exposable configuration settings.
-
-        A more complete list is returned if the user is an admin.
-        Pass in `view` and a comma-seperated list of keys to control which
-        configuration settings are returned.
-        """
-        return _index(self.configuration_manager, trans, view, keys)
-
-    @router.get(
-        "/api/version",
-        summary="Return Galaxy version information: major/minor version, optional extra info",
-        response_description="Galaxy version information: major/minor version, optional extra info",
-    )
-    def version(self) -> Dict[str, Any]:
-        """Return Galaxy version information: major/minor version, optional extra info."""
-        return self.configuration_manager.version()
-
-    @router.get(
-        "/api/configuration/dynamic_tool_confs",
-        require_admin=True,
-        summary="Return dynamic tool configuration files",
-        response_description="Dynamic tool configuration files",
-    )
-    def dynamic_tool_confs(self) -> List[Dict[str, str]]:
-        """Return dynamic tool configuration files."""
-        return self.configuration_manager.dynamic_tool_confs()
-
-    @router.get(
-        "/api/configuration/decode/{encoded_id}",
-        require_admin=True,
-        summary="Decode a given id",
-        response_description="Decoded id",
-    )
-    def decode_id(self, encoded_id: str = EncodedIdPathParam) -> Dict[str, int]:
-        """Decode a given id."""
-        return self.configuration_manager.decode_id(encoded_id)
-
-    @router.get(
-        "/api/configuration/tool_lineages",
-        require_admin=True,
-        summary="Return tool lineages for tools that have them",
-        response_description="Tool lineages for tools that have them",
-    )
-    def tool_lineages(self) -> List[Dict[str, Dict]]:
-        """Return tool lineages for tools that have them."""
-        return self.configuration_manager.tool_lineages()
-
-    @router.put(
-        "/api/configuration/toolbox", require_admin=True, summary="Reload the Galaxy toolbox (but not individual tools)"
-    )
-    def reload_toolbox(self):
-        """Reload the Galaxy toolbox (but not individual tools)."""
-        self.configuration_manager.reload_toolbox()
+@router.get(
+    "/api/whoami",
+    summary="Return information about the current authenticated user",
+    response_description="Information about the current authenticated user",
+)
+def whoami(trans: ProvidesUserContext = DependsOnTrans) -> Optional[UserModel]:
+    """Return information about the current authenticated user."""
+    return _user_to_model(trans.user)
+
+
+@router.get(
+    "/api/configuration",
+    summary="Return an object containing exposable configuration settings",
+    response_description="Object containing exposable configuration settings",
+)
+def index(
+    trans: ProvidesUserContext = DependsOnTrans,
+    view: Optional[str] = SerializationViewQueryParam,
+    keys: Optional[str] = SerializationKeysQueryParam,
+    configuration_manager: ConfigurationManager = depends(ConfigurationManager),
+) -> Dict[str, Any]:
+    """
+    Return an object containing exposable configuration settings.
+
+    A more complete list is returned if the user is an admin.
+    Pass in `view` and a comma-seperated list of keys to control which
+    configuration settings are returned.
+    """
+    return _index(configuration_manager, trans, view, keys)
+
+
+@router.get(
+    "/api/version",
+    summary="Return Galaxy version information: major/minor version, optional extra info",
+    response_description="Galaxy version information: major/minor version, optional extra info",
+)
+def version(
+    configuration_manager: ConfigurationManager = depends(ConfigurationManager),
+) -> Dict[str, Any]:
+    """Return Galaxy version information: major/minor version, optional extra info."""
+    return configuration_manager.version()
+
+
+@router.get(
+    "/api/configuration/dynamic_tool_confs",
+    require_admin=True,
+    summary="Return dynamic tool configuration files",
+    response_description="Dynamic tool configuration files",
+)
+def dynamic_tool_confs(
+    configuration_manager: ConfigurationManager = depends(ConfigurationManager),
+) -> List[Dict[str, str]]:
+    """Return dynamic tool configuration files."""
+    return configuration_manager.dynamic_tool_confs()
+
+
+@router.get(
+    "/api/configuration/decode/{encoded_id}",
+    require_admin=True,
+    summary="Decode a given id",
+    response_description="Decoded id",
+)
+def decode_id(
+    encoded_id: str = EncodedIdPathParam,
+    configuration_manager: ConfigurationManager = depends(ConfigurationManager),
+) -> Dict[str, int]:
+    """Decode a given id."""
+    return configuration_manager.decode_id(encoded_id)
+
+
+@router.get(
+    "/api/configuration/tool_lineages",
+    require_admin=True,
+    summary="Return tool lineages for tools that have them",
+    response_description="Tool lineages for tools that have them",
+)
+def tool_lineages(
+    configuration_manager: ConfigurationManager = depends(ConfigurationManager),
+) -> List[Dict[str, Dict]]:
+    """Return tool lineages for tools that have them."""
+    return configuration_manager.tool_lineages()
+
+
+@router.put(
+    "/api/configuration/toolbox", require_admin=True, summary="Reload the Galaxy toolbox (but not individual tools)"
+)
+def reload_toolbox(
+    configuration_manager: ConfigurationManager = depends(ConfigurationManager),
+):
+    """Reload the Galaxy toolbox (but not individual tools)."""
+    configuration_manager.reload_toolbox()
 
 
 def _user_to_model(user):