devx/src/devx/tools/clean_images.py

#!/usr/bin/env python3
"""Clean up old Docker images from a Gitea container registry.

Queries the Gitea API for all versions of a package (container type) and
deletes all but the most recent N versions.  The ``latest`` tag is always
preserved if present.

.. note::
    This tool only deletes package versions via the Gitea API. The underlying
    blob files on the Gitea server's filesystem are NOT removed by this tool
    (Gitea 1.26.x has no built-in garbage collection). The production VM's
    daily cleanup script (``cleanup_gitea.py``) handles filesystem blob GC
    by querying the database for referenced blobs and removing orphaned files.

Usage::

    # Clean up ci-base images, keep last 2 versions
    python3 -m devx.tools.clean_images \\
        --owner oblachno-oss \\
        --name ci-base \\
        --keep 2

    # Clean up multiple images
    python3 -m devx.tools.clean_images \\
        --owner oblachno-oss \\
        --name ci-base \\
        --name ci-quality \\
        --name ci-full \\
        --keep 2

    # Dry run (list what would be deleted)
    python3 -m devx.tools.clean_images \\
        --owner oblachno-oss \\
        --name ci-base \\
        --keep 2 \\
        --dry-run

Authentication uses ``CI_GITEA_API_TOKEN`` environment variable (or legacy ``CI_GITEA_TOKEN``).
"""

from __future__ import annotations

import time
from typing import Any

import click
import requests

from devx.config import GITEA_API_URL, REPO_OWNER
from devx.i18n import _
from devx.tokens import get_developer_token


def list_package_versions(
    api_url: str,
    owner: str,
    name: str,
    token: str,
    *,
    timeout: int = 30,
) -> list[dict[str, Any]]:
    """List all versions of a container package from the Gitea API.

    Returns a list of version dicts, each containing at least ``version``
    and ``created_at`` fields.
    """
    from urllib.parse import quote

    encoded_name = quote(name, safe="")
    url = f"{api_url}/packages/{owner}?type=container&name={encoded_name}"
    headers = {"Authorization": f"token {token}"}
    all_versions: list[dict[str, Any]] = []
    page = 1
    while True:
        resp = requests.get(
            f"{url}&page={page}&limit=50",
            headers=headers,
            timeout=timeout,
        )
        resp.raise_for_status()
        data = resp.json()
        if not data:
            break
        all_versions.extend(data)
        if len(data) < 50:
            break
        page += 1
    return all_versions


def delete_package_version(
    api_url: str,
    owner: str,
    name: str,
    version: str,
    token: str,
    *,
    timeout: int = 30,
    package_type: str = "container",
    max_retries: int = 3,
) -> bool:
    """Delete a specific version of a container package.

    Uses the Gitea API endpoint ``DELETE /packages/{owner}/{type}/{name}/{version}``.
    Retries on transient failures (5xx, timeouts) up to ``max_retries`` times.

    Returns True on success, False on failure.
    """
    from urllib.parse import quote

    encoded_name = quote(name, safe="")
    encoded_version = quote(version, safe="")
    url = f"{api_url}/packages/{owner}/{package_type}/{encoded_name}/{encoded_version}"
    headers = {"Authorization": f"token {token}"}
    for attempt in range(max_retries):
        try:
            resp = requests.delete(url, headers=headers, timeout=timeout)
        except requests.RequestException:
            if attempt < max_retries - 1:
                time.sleep(2**attempt)
                continue
            return False
        if resp.status_code in (204, 200):
            return True
        # 404 means already deleted — treat as success
        if resp.status_code == 404:
            return True
        # 5xx is transient — retry
        if 500 <= resp.status_code < 600 and attempt < max_retries - 1:
            time.sleep(2**attempt)
            continue
        return False
    return False


def sort_versions_by_date(
    versions: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Sort package versions by creation date, newest first.

    Falls back to version string comparison if created_at is missing.
    """

    def _sort_key(v: dict[str, Any]) -> str:
        return str(v.get("created_at", v.get("version", "")))

    return sorted(versions, key=_sort_key, reverse=True)


def select_for_deletion(
    versions: list[dict[str, Any]],
    keep: int,
) -> list[dict[str, Any]]:
    """Select versions to delete, keeping the most recent ``keep`` versions.

    Versions named ``latest`` are always preserved.
    """
    sorted_versions = sort_versions_by_date(versions)
    to_delete = sorted_versions[keep:]
    # Always preserve 'latest' tag
    to_delete = [v for v in to_delete if v.get("version") != "latest"]
    return to_delete


@click.command()
@click.option(
    "--owner",
    default=None,
    help="Package owner (user or org, default: from [tool.devx] repo_owner).",
)
@click.option(
    "--name",
    "names",
    multiple=True,
    required=True,
    help="Package name(s). Can be repeated.",
)
@click.option(
    "--keep",
    default=2,
    type=int,
    show_default=True,
    help="Number of recent versions to keep (excluding 'latest').",
)
@click.option(
    "--dry-run",
    is_flag=True,
    default=False,
    help="List versions that would be deleted without actually deleting.",
)
@click.option(
    "--api-url",
    default=None,
    help="Gitea API URL (defaults to DEVX_GITEA_API_URL or built-in default).",
)
def main(
    owner: str | None,
    names: tuple[str, ...],
    keep: int,
    dry_run: bool,
    api_url: str | None,
) -> None:
    """Clean up old Docker image versions from a Gitea registry."""
    try:
        token = get_developer_token()
    except click.ClickException:
        raise click.ClickException(_("CI_GITEA_TOKEN environment variable required")) from None
    if not owner:
        owner = REPO_OWNER
    if not owner:
        raise click.ClickException(_("Package owner not specified. Use --owner or set [tool.devx] repo_owner."))
    base_url = api_url or GITEA_API_URL

    total_deleted = 0
    total_kept = 0
    total_failed = 0
    for name in names:
        click.echo(_("\n{separator}", separator="=" * 60))
        click.echo(_("Package: {owner}/{name}", owner=owner, name=name))
        click.echo(_("{separator}", separator="=" * 60))
        try:
            versions = list_package_versions(base_url, owner, name, token)
        except requests.RequestException as exc:
            click.echo(
                _("Failed to list versions for {name}: {error}", name=name, error=exc),
                err=True,
            )
            total_failed += 1
            continue

        if not versions:
            click.echo(_("No versions found."))
            continue

        click.echo(_("Found {count} version(s):", count=len(versions)))
        for v in sort_versions_by_date(versions):
            click.echo(
                _("  {version} (created: {created})", version=v.get("version", "?"), created=v.get("created_at", "?"))
            )

        to_delete = select_for_deletion(versions, keep)
        kept_count = len(versions) - len(to_delete)
        click.echo(_("\nKeeping {kept}, would delete {count}", kept=kept_count, count=len(to_delete)))

        if dry_run:
            for v in to_delete:
                click.echo(_("  [dry-run] Would delete: {version}", version=v.get("version", "?")))
            total_kept += kept_count
            continue

        deleted_count = 0
        failed_count = 0
        for v in to_delete:
            version = str(v.get("version", ""))
            if delete_package_version(base_url, owner, name, version, token):
                click.echo(_("  Deleted: {version}", version=version))
                deleted_count += 1
            else:
                click.echo(_("  FAILED to delete: {version}", version=version), err=True)
                failed_count += 1

        total_deleted += deleted_count
        total_kept += kept_count
        total_failed += failed_count

    click.echo(
        _(
            "\nDone. Deleted {deleted}, kept {kept}, failed {failed}.",
            deleted=total_deleted,
            kept=total_kept,
            failed=total_failed,
        )
    )
    if total_failed > 0:
        raise click.ClickException(_("Failed to delete {count} image version(s)", count=total_failed))


if __name__ == "__main__":  # pragma: no cover
    main()  # pragma: no cover