Merge pull request #26 from ricardogsilva/18-use-typer-for-ci-commands

use typer for ci commands

.github/workflows/ci.yaml

@@ -26,14 +26,17 @@ jobs:
       - name: setup Python
         uses: actions/setup-python@v5
         with:
-          python-version: '3.10'
+          python-version: "3.10"
           cache: pip
           cache-dependency-path: docker/backend/project_requirements.txt
 
-      - name: install dagger for python
-        uses: insightsengineering/pip-action@v2
+      - name: setup poetry
+        uses: Gr1N/setup-poetry@v9
         with:
-          packages: dagger-io==0.9.8
+          poetry-version: "1.7.1"
+
+      - name: install code with dev dependencies
+        run: poetry install --with dev
 
       - name: login to container registry
         if: ${{ env.PUBLISH_IMAGE == 'TRUE' }}
@@ -48,7 +51,7 @@ jobs:
         with:
           verb: run
           args: >-
-            python tests/ci/main.py 
+            poetry run python tests/ci/main.py 
             --with-tests 
             ${{ env.PUBLISH_IMAGE == 'TRUE' && format('--publish-docker-image {0}:{1}', env.IMAGE_NAME, env.IMAGE_TAG)  || ''}}
           version: 0.9.9
@@ -59,5 +62,5 @@ jobs:
         uses: dagger/dagger-for-github@v5
         with:
           verb: run
-          args: python tests/ci/main.py --with-security-scan
+          args: poetry run python tests/ci/main.py --with-security-scan
           version: 0.9.9

tests/ci/main.py

@@ -1,18 +1,58 @@
-import argparse
 import asyncio
 import logging
 import os
 import shlex
 import sys
 from pathlib import Path
+from typing import Annotated
 
 import dagger
+import typer
 
 logger = logging.getLogger(__name__)
 POSTGIS_IMAGE_VERSION = "postgis/postgis:16-3.4"
 
+cli_app = typer.Typer()
 
-def sanitize_docker_image_name(docker_image_name: str) -> str:
+
+@cli_app.command()
+def run_ci_pipeline(
+    with_tests: Annotated[
+            bool,
+            typer.Option(
+                help=(
+                        "Run automated tests on the built container and exit with an "
+                        "error if a test fails."
+                )
+            )
+        ] = False,
+        with_security_scan: Annotated[
+            bool,
+            typer.Option(
+                help=(
+                        "Full URI to an image registry where the built container image should be "
+                        "published, including the image tag. This assumes that logging in to the "
+                        "registry has already been made (for example by running the "
+                        "`docker login` command beforehand)."
+                        "Example: ghcr.io/geobeyond/arpav-ppcv-backend:latest"
+                )
+            )
+        ] = False,
+        publish_docker_image: str | None = None
+):
+    """Command-line interface for running CI pipeline."""
+
+    logging.basicConfig(level=logging.INFO)
+    return asyncio.run(
+        _run_pipeline(
+            with_tests=with_tests,
+            with_security_scan=with_security_scan,
+            publish_docker_image=publish_docker_image
+        )
+    )
+
+
+def _sanitize_docker_image_name(docker_image_name: str) -> str:
     """Ensure input docker_image_name is valid.
 
     This function sanitizes the input according to the rules described in
@@ -32,7 +72,7 @@ def sanitize_docker_image_name(docker_image_name: str) -> str:
     return f"{host}/{path.lower()}:{tag or 'latest'}"
 
 
-def get_env_variables() -> dict[str, str | None]:
+def _get_env_variables() -> dict[str, str | None]:
     return {
         "DEBUG": os.getenv("DEBUG", "0"),
         "PGPASSWORD": os.getenv("PGPASSWORD", "postgres"),
@@ -50,7 +90,7 @@ def get_env_variables() -> dict[str, str | None]:
     }
 
 
-async def run_security_scan(built_container: dagger.Container):
+async def _run_security_scan(built_container: dagger.Container):
     return await (
         built_container.with_user("root")
         .without_entrypoint()
@@ -77,7 +117,7 @@ async def run_security_scan(built_container: dagger.Container):
     ).stdout()
 
 
-async def run_tests(
+async def _run_tests(
         client: dagger.Client,
         built_container: dagger.Container,
         env_variables: dict[str, str]
@@ -125,13 +165,13 @@ async def run_tests(
     ).stdout()
 
 
-async def run_pipeline(
+async def _run_pipeline(
         *,
         with_tests: bool,
         with_security_scan: bool,
         publish_docker_image: str | None = None
 ):
-    env_variables = get_env_variables()
+    env_variables = _get_env_variables()
     conf = dagger.Config(
         log_output=sys.stderr,
     )
@@ -150,51 +190,14 @@ async def run_pipeline(
             )
         )
         if with_security_scan:
-            await run_security_scan(built_container)
+            await _run_security_scan(built_container)
         if with_tests:
-            await run_tests(client, built_container, env_variables)
+            await _run_tests(client, built_container, env_variables)
         if publish_docker_image is not None:
-            sanitized_name = sanitize_docker_image_name(publish_docker_image)
+            sanitized_name = _sanitize_docker_image_name(publish_docker_image)
             await built_container.publish(sanitized_name)
-
         print("Done")
 
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser()
-    parser.add_argument(
-        "--with-security-scan",
-        action="store_true",
-        help=(
-            "Run the trivy security scanner on the built container image in order "
-            "to find known vulnerabilities of level HIGH and CRITICAL. Exits with "
-            "an error if any vulnerabilities are found."
-        )
-    )
-    parser.add_argument(
-        "--with-tests",
-        action="store_true",
-        help=(
-            "Run automated tests on the built container and exit with an error if a "
-            "test fails."
-        )
-    )
-    parser.add_argument(
-        "--publish-docker-image",
-        help=(
-            "Full URI to an image registry where the built container image should be "
-            "published, including the image tag. This assumes that logging in to the "
-            "registry has already been made (for example by running the "
-            "`docker login` command beforehand)."
-            "Example: ghcr.io/geobeyond/arpav-ppcv-backend:latest"
-        )
-    )
-    args = parser.parse_args()
-    logging.basicConfig(level=logging.INFO)
-    asyncio.run(
-        run_pipeline(
-            with_tests=args.with_tests,
-            with_security_scan=args.with_security_scan,
-            publish_docker_image=args.publish_docker_image,
-        )
-    )
+    cli_app()