Merge branch 'api-delay-for-dry-run' into 'main'

from datetime import datetime, timedelta

from enum import Enum

from fnmatch import fnmatch

from typing import Optional

from typing import Iterator, Optional

import yaml

from gitlab import (

        # return True if pipeline was actually deleted

        return deleted

    def project_pipelines_iterator(self, project: Project) -> Iterator[ProjectPipeline]:

        """

        List all pipelines of a project. Using this method helps for:

        - introducing a slight delay between API calls for next results page, if in dry-run mode

          (because in this case we iterate quickly)

        - mypy type checking, because list() methods in python-gitlab are not narrowly typed

        :param project:

        :return: a ProjectPipeline iterator

        """

        iterator = project.pipelines.list(iterator=True, sort="asc", order_by="updated_at")

        assert isinstance(iterator, Iterator) # help mypy

        yield_count = 0

        while True:

            try:

                # maybe sleep if we are about to request the next pipelines page

                if yield_count and self.dry_run and self.delay_between_api_call and (yield_count % (self.client.per_page or 20) == 0):

                    time.sleep(self.delay_between_api_call)

                pipeline = next(iterator)

                assert isinstance(pipeline, ProjectPipeline) # help mypy

                yield pipeline

                yield_count += 1

            except StopIteration:

                break

    # Cleanup a GitLab project

    # $1: project to be cleaned

    def clean_project(self, project: Project) -> None:

        kept_pipelines: dict[str, list[ProjectPipeline]] = {}

        # iterate over project pipelines (starting with oldest)

        for pipeline in project.pipelines.list(iterator=True, sort="asc", order_by="updated_at"):

        for pipeline in self.project_pipelines_iterator(project):

            # stop right there if enough pipelines deletions have been attempted already

            if self.pipeline_deletion_limit and (pipelines_deletions_attempts >= self.pipeline_deletion_limit):

                if self.verbose:

                    print('Pipeline deletion limit reached, stopping processing for this project.')

                break

            # list() methods in python-gitlab are not narrowly typed, so let's help mypy

            assert isinstance(pipeline, ProjectPipeline)

            # if the pipeline has been updated after the cutoff date: keep and break loop

            if datetime.strptime(pipeline.updated_at, "%Y-%m-%dT%H:%M:%S.%fZ") >= updated_limit:

                if self.verbose and self.debug: