Fastapi migration p1

packit_service/models.py

@@ -12,6 +12,7 @@
 from collections.abc import Generator, Iterable
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
+from http import HTTPStatus
 from os import getenv
 from typing import (
     TYPE_CHECKING,
@@ -24,6 +25,7 @@
 from cachetools import TTLCache, cached
 from cachetools.func import ttl_cache
 from packit.config import JobConfigTriggerType
+from pydantic import BaseModel
 from sqlalchemy import (
     JSON,
     Boolean,
@@ -4648,3 +4650,8 @@ def get_onboarded_projects() -> tuple[dict[int, str], dict[int, str]]:
         for project in recheck_if_onboarded.difference(onboarded_projects)
     }
     return (onboarded, almost_onboarded)
+
+
+class BodhiUpdatesListResponse(BaseModel):
+    result: List
+    status: HTTPStatus

packit_service/service/api/bodhi_updates.py

@@ -4,114 +4,128 @@
 from http import HTTPStatus
 from logging import getLogger
 
-from flask_restx import Namespace, Resource
+from fastapi import APIRouter, Depends, Path, Response
+from fastapi.exceptions import HTTPException
+from parsers import Pagination_Arguments
+from response_models import (
+    BodhiUpdateGroupResponse,
+    BodhiUpdateItemResponse,
+    BodhiUpdatesListResponse,
+)
 
 from packit_service.models import (
     BodhiUpdateGroupModel,
     BodhiUpdateTargetModel,
     optional_timestamp,
 )
-from packit_service.service.api.parsers import indices, pagination_arguments
-from packit_service.service.api.utils import get_project_info_from_build, response_maker
+from packit_service.service.api.parsers import indices
+from packit_service.service.api.utils import get_project_info_from_build
 
 logger = getLogger("packit_service")
 
-ns = Namespace("bodhi-updates", description="Bodhi updates")
-
-
-@ns.route("")
-class BodhiUpdatesList(Resource):
-    @ns.expect(pagination_arguments)
-    @ns.response(HTTPStatus.PARTIAL_CONTENT, "Bodhi updates list follows")
-    def get(self):
-        """List all Bodhi updates."""
-        first, last = indices()
-        result = []
-
-        for update in BodhiUpdateTargetModel.get_range(first, last):
-            update_dict = {
-                "packit_id": update.id,
-                "status": update.status,
-                "branch": update.target,
-                "web_url": update.web_url,
-                "koji_nvrs": update.koji_nvrs,
-                "alias": update.alias,
-                "pr_id": update.get_pr_id(),
-                "branch_name": update.get_branch_name(),
-                "release": update.get_release_tag(),
-                "submitted_time": optional_timestamp(update.submitted_time),
-                "update_creation_time": optional_timestamp(update.update_creation_time),
-            }
-
-            if project := update.get_project():
-                update_dict["project_url"] = project.project_url
-                update_dict["repo_namespace"] = project.namespace
-                update_dict["repo_name"] = project.repo_name
-
-            result.append(update_dict)
-
-        resp = response_maker(
-            result,
-            status=HTTPStatus.PARTIAL_CONTENT,
-        )
-        resp.headers["Content-Range"] = f"bodhi-updates {first + 1}-{last}/*"
-        return resp
-
+# ns = Namespace("bodhi-updates", description="Bodhi updates")
+router: APIRouter = APIRouter(prefix="/api/bodhi-updates", tags=["bodhi-updates"])
 
-@ns.route("/<int:id>")
-@ns.param("id", "Packit id of the update")
-class BodhiUpdateItem(Resource):
-    @ns.response(HTTPStatus.OK, "OK, Bodhi update details follow")
-    @ns.response(HTTPStatus.NOT_FOUND.value, "No info about Bodhi update stored in DB")
-    def get(self, id):
-        """A specific Bodhi updates details."""
-        update = BodhiUpdateTargetModel.get_by_id(int(id))
 
-        if not update:
-            return response_maker(
-                {"error": "No info about update stored in DB"},
-                status=HTTPStatus.NOT_FOUND,
-            )
+# @ns.expect(pagination_arguments)
+# @ns.response(HTTPStatus.PARTIAL_CONTENT, "Bodhi updates list follows")
+@router.get("/", response_model=BodhiUpdatesListResponse)
+def BodhiUpdatesList(
+    response: Response, params: Pagination_Arguments = Depends()
+) -> BodhiUpdatesListResponse:
+    """List all Bodhi updates."""
+    first, last = indices(params)
+    result = []
 
+    for update in BodhiUpdateTargetModel.get_range(first, last):
         update_dict = {
+            "packit_id": update.id,
             "status": update.status,
             "branch": update.target,
             "web_url": update.web_url,
             "koji_nvrs": update.koji_nvrs,
             "alias": update.alias,
+            "pr_id": update.get_pr_id(),
+            "branch_name": update.get_branch_name(),
+            "release": update.get_release_tag(),
             "submitted_time": optional_timestamp(update.submitted_time),
             "update_creation_time": optional_timestamp(update.update_creation_time),
-            "run_ids": sorted(run.id for run in update.group_of_targets.runs),
-            "error_message": update.data.get("error") if update.data else None,
         }
 
-        update_dict.update(get_project_info_from_build(update))
-        return response_maker(update_dict)
-
-
-@ns.route("/groups/<int:id>")
-@ns.param("id", "Packit id of the Bodhi update group")
-class BodhiUpdateGroup(Resource):
-    @ns.response(HTTPStatus.OK, "OK, Bodhi update group details follow")
-    @ns.response(
-        HTTPStatus.NOT_FOUND.value,
-        "No info about Bodhi update group stored in DB",
-    )
-    def get(self, id):
-        """A specific Bodhi update group details."""
-        group_model = BodhiUpdateGroupModel.get_by_id(int(id))
-
-        if not group_model:
-            return response_maker(
-                {"error": "No info about group stored in DB"},
-                status=HTTPStatus.NOT_FOUND,
-            )
-
-        group_dict = {
-            "submitted_time": optional_timestamp(group_model.submitted_time),
-            "run_ids": sorted(run.id for run in group_model.runs),
-            "update_target_ids": sorted(build.id for build in group_model.grouped_targets),
-        }
+        if project := update.get_project():
+            update_dict["project_url"] = project.project_url
+            update_dict["repo_namespace"] = project.namespace
+            update_dict["repo_name"] = project.repo_name
+
+        result.append(update_dict)
+
+    response.headers["Content-Range"] = f"bodhi-updates {first + 1}-{last}/*"
+    return BodhiUpdatesListResponse(result)
+
+
+# @ns.route("/<int:id>")
+# @ns.param("id", "Packit id of the update")
+# class BodhiUpdateItem(Resource):
+# @ns.response(HTTPStatus.OK, "OK, Bodhi update details follow")
+# @ns.response(HTTPStatus.NOT_FOUND.value, "No info about Bodhi update stored in DB")
+@router.get(
+    "/{id}",
+    response_model=BodhiUpdateItemResponse,
+    responses={HTTPStatus.NOT_FOUND: {"description": "No info about update stored in DB"}},
+)
+def BodhiUpdateItem(id: int = Path(..., description="Packit id of the update")):
+    """A specific Bodhi updates details."""
+    update = BodhiUpdateTargetModel.get_by_id(int(id))
+
+    if not update:
+        raise HTTPException(
+            detail="No info about update stored in DB", status_code=HTTPStatus.NOT_FOUND
+        )
+
+    update_dict = {
+        "status": update.status,
+        "branch": update.target,
+        "web_url": update.web_url,
+        "koji_nvrs": update.koji_nvrs,
+        "alias": update.alias,
+        "submitted_time": optional_timestamp(update.submitted_time),
+        "update_creation_time": optional_timestamp(update.update_creation_time),
+        "run_ids": sorted(run.id for run in update.group_of_targets.runs),
+        "error_message": update.data.get("error") if update.data else None,
+    }
+
+    update_dict.update(get_project_info_from_build(update))
+    return BodhiUpdateItemResponse(update_dict)
+
+
+# @ns.route("/groups/<int:id>")
+# @ns.param("id", "Packit id of the Bodhi update group")
+# class BodhiUpdateGroup(Resource):
+# @ns.response(HTTPStatus.OK, "OK, Bodhi update group details follow")
+# @ns.response(
+#     HTTPStatus.NOT_FOUND.value,
+#     "No info about Bodhi update group stored in DB",
+# )
+@router.get(
+    "/groups/{id}",
+    response_model=BodhiUpdateGroupResponse,
+    responses={HTTPStatus.NOT_FOUND: {"description": "No info about group stored in DB"}},
+)
+def BodhiUpdateGroup(id: int = Path(..., description="Packit id of the Bodhi update group")):
+    """A specific Bodhi update group details."""
+    group_model = BodhiUpdateGroupModel.get_by_id(int(id))
+
+    if not group_model:
+        raise HTTPException(
+            detail="No info about group stored in DB",
+            status=HTTPStatus.NOT_FOUND,
+        )
+
+    group_dict = {
+        "submitted_time": optional_timestamp(group_model.submitted_time),
+        "run_ids": sorted(run.id for run in group_model.runs),
+        "update_target_ids": sorted(build.id for build in group_model.grouped_targets),
+    }
 
-        group_dict.update(get_project_info_from_build(group_model))
-        return response_maker(group_dict)
+    group_dict.update(get_project_info_from_build(group_model))
+    return BodhiUpdateGroupResponse(group_dict)

packit_service/service/api/parsers.py

@@ -1,37 +1,53 @@
 # Copyright Contributors to the Packit project.
 # SPDX-License-Identifier: MIT
 
-from flask import request
-from flask_restx import reqparse
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field
 
 DEFAULT_PAGE = 1
 DEFAULT_PER_PAGE = 10
 
-pagination_arguments = reqparse.RequestParser()
-pagination_arguments.add_argument(
-    "page",
-    type=int,
-    required=False,
-    default=1,
-    help="Page number",
-)
-pagination_arguments.add_argument(
-    "per_page",
-    type=int,
-    required=False,
-    choices=[2, 10, 20, 30, 40, 50],
-    default=DEFAULT_PER_PAGE,
-    help="Results per page",
-)
-
-
-def indices():
+
+class PerPageChoices(int, Enum):
+    TWO = 2
+    TEN = 10
+    TWENTY = 20
+    THIRTY = 30
+    FORTY = 40
+    FIFTY = 50
+
+
+# pagination_arguments = reqparse.RequestParser()
+# pagination_arguments.add_argument(
+#     "page",
+#     type=int,
+#     required=False,
+#     default=1,
+#     help="Page number",
+# )
+# pagination_arguments.add_argument(
+#     "per_page",
+#     type=int,
+#     required=False,
+#     choices=[2, 10, 20, 30, 40, 50],
+#     default=DEFAULT_PER_PAGE,
+#     help="Results per page",
+# )
+
+
+class Pagination_Arguments(BaseModel):
+    page: Optional[int] = Field(default=1, description="Page number")
+    per_page: Optional[PerPageChoices] = Field(
+        default=DEFAULT_PER_PAGE, description="Results per page"
+    )
+
+
+def indices(pagination_arguments: Pagination_Arguments):
     """Return indices of first and last entry based on request arguments"""
-    args = pagination_arguments.parse_args(request)
-    page = args.get("page", DEFAULT_PAGE)
-    if page < DEFAULT_PAGE:
-        page = DEFAULT_PAGE
-    per_page = args.get("per_page", DEFAULT_PER_PAGE)
+    page = pagination_arguments.page
+    per_page = pagination_arguments.per_page
     first = (page - 1) * per_page
     last = page * per_page
     return first, last

packit_service/service/api/response_models.py

@@ -0,0 +1,18 @@
+# Copyright Contributors to the Packit project.
+# SPDX-License-Identifier: MIT
+
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel
+
+
+class BodhiUpdatesListResponse(BaseModel):
+    result: Optional[List]
+
+
+class BodhiUpdateItemResponse(BaseModel):
+    update_dict: Dict[str, Any]
+
+
+class BodhiUpdateGroupResponse(BaseModel):
+    group_dict: Dict[str, Any]