Convert all classes w/methods into functions

Although this would work, we'd need to add the staticmethod decorator to make it right (and fix mypy). Functions, I think, are simpler and cleaner here.
jdavcs · Dec 16, 2023 · f897c8d · f897c8d
1 parent 14113e3
commit f897c8d
Show file tree

Hide file tree

Showing 36 changed files with 2,831 additions and 2,723 deletions.
diff --git a/lib/galaxy/webapps/galaxy/api/authenticate.py b/lib/galaxy/webapps/galaxy/api/authenticate.py
@@ -48,14 +48,13 @@ def options(self, trans: GalaxyWebTransaction, **kwd):
         # trans.response.headers['Access-Control-Allow-Methods'] = 'POST, PUT, GET, OPTIONS, DELETE'
 
 
-class FastAPIAuthenticate:
-    @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)
+@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)
diff --git a/lib/galaxy/webapps/galaxy/api/cloud.py b/lib/galaxy/webapps/galaxy/api/cloud.py
@@ -32,68 +32,69 @@
 router = Router(tags=["cloud"])
 
 
-class FastAPICloudController:
-    @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(
+    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,
-    )
-    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,
-        )
-        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,
+@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 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,
-            )
-        )
+    rtv = []
+    for dataset in datasets:
+        rtv.append(datasets_serializer.serialize_to_view(dataset, view="summary"))
+    return DatasetSummaryList.construct(__root__=rtv)
 
-        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,
+
+@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)
diff --git a/lib/galaxy/webapps/galaxy/api/configuration.py b/lib/galaxy/webapps/galaxy/api/configuration.py
@@ -39,92 +39,97 @@
 )
 
 
-class FastAPIConfiguration:
-    @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()
+@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):