feat: check if build exists already

Signed-off-by: Jyotiraditya Panda <jyotiraditya@aospa.co>

dumpyarabot/handlers.py

@@ -38,11 +38,36 @@ async def dump_main(
         )
         return
 
-    # Try to call jenkins
+    # Try to check for existing build and call jenkins if necessary
     try:
-        response_text = await utils.call_jenkins(
-            schemas.DumpArguments(url=context.args[0], use_alt_dumper=use_alt_dumper)
+        dump_args = schemas.DumpArguments(
+            url=context.args[0], use_alt_dumper=use_alt_dumper
         )
+
+        # Check for existing build
+        initial_message = await context.bot.send_message(
+            chat_id=update.effective_chat.id,
+            reply_to_message_id=update.effective_message.id,
+            text="Checking for existing builds...",
+        )
+
+        exists, message = await utils.check_existing_build(dump_args)
+        await context.bot.edit_message_text(
+            chat_id=update.effective_chat.id,
+            message_id=initial_message.message_id,
+            text=f"{message}\n\n",
+        )
+
+        if exists:
+            return
+
+        await context.bot.delete_message(
+            chat_id=update.effective_chat.id,
+            message_id=initial_message.message_id,
+        )
+
+        # If no existing build, call jenkins
+        response_text = await utils.call_jenkins(dump_args)
     except ValidationError:
         response_text = "Invalid URL"
     except Exception:

dumpyarabot/utils.py

@@ -1,9 +1,102 @@
+from typing import Dict, List, Tuple
+
 import httpx
 
 from dumpyarabot import schemas
 from dumpyarabot.config import settings
 
 
+async def get_jenkins_builds() -> List[Dict]:
+    """
+    Fetch all builds from Jenkins.
+
+    Returns:
+        List[Dict]: A list of dictionaries containing build information.
+
+    Raises:
+        httpx.HTTPStatusError: If the API request fails.
+    """
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            f"{settings.JENKINS_URL}/job/dumpyara/api/json",
+            params={"tree": "allBuilds[number,result,actions[parameters[name,value]]]"},
+        )
+        response.raise_for_status()
+        return response.json()["allBuilds"]
+
+
+async def check_existing_build(args: schemas.DumpArguments) -> Tuple[bool, str]:
+    """
+    Check if a build with the given parameters already exists.
+
+    Args:
+        args (schemas.DumpArguments): The schema for the Jenkins call.
+
+    Returns:
+        Tuple[bool, str]: A tuple where the boolean indicates if a matching build was found,
+                          and the string provides additional information about the build status.
+    """
+    builds = await get_jenkins_builds()
+
+    for build in builds:
+        if _is_matching_build(build, args):
+            return _get_build_status(build)
+
+    return False, "No matching build found. A new build will be started."
+
+
+def _is_matching_build(build: Dict, args: schemas.DumpArguments) -> bool:
+    """
+    Check if a build matches the given arguments.
+
+    Args:
+        build (Dict): The build information.
+        args (schemas.DumpArguments): The schema for the Jenkins call.
+
+    Returns:
+        bool: True if the build matches the arguments, False otherwise.
+    """
+    for action in build.get("actions", []):
+        if "parameters" in action:
+            params = {param["name"]: param["value"] for param in action["parameters"]}
+            return (
+                params.get("URL") == args.url.unicode_string()
+                and params.get("USE_ALT_DUMPER") == args.use_alt_dumper
+            )
+    return False
+
+
+def _get_build_status(build: Dict) -> Tuple[bool, str]:
+    """
+    Get the status of a build.
+
+    Args:
+        build (Dict): The build information.
+
+    Returns:
+        Tuple[bool, str]: A tuple where the boolean indicates if a new build should be started,
+                          and the string provides information about the build status.
+    """
+    result = build.get("result")
+    build_number = build["number"]
+
+    if result is None:
+        return (
+            True,
+            f"Build #{build_number} is currently in progress for this URL and settings.",
+        )
+    elif result == "SUCCESS":
+        return (
+            True,
+            f"Build #{build_number} has already successfully completed for this URL and settings.",
+        )
+    else:
+        return (
+            False,
+            f"Build #{build_number} exists for this URL and settings, but result was {result}. A new build will be started.",
+        )
+
+
 async def call_jenkins(args: schemas.DumpArguments) -> str:
     """
     Function to call jenkins