style: fix formatting issues (dbcfcbc0) · Commits · to be continuous... / tools / gitlab-butler

gitlab_butler/butler.py

+152 −90

Original line number	Diff line number	Diff line
		@@ -23,28 +23,18 @@ class PipelineSourceEnum(Enum):
		TAG = 2
		MERGE_REQUEST = 3


		# Cache to avoid requesting branch/tags/mr statuses multiple times
		class ProjectSourcesCache:

		class Branch:
		def __init__(
		self,
		name: str,
		protected: bool,
		default: bool
		) -> None:
		def __init__(self, name: str, protected: bool, default: bool) -> None:
		self.name: str = name
		self.protected: bool = protected
		self.default: bool = default

		class MergeRequest:
		def __init__(
		self,
		id: str,
		iid: str,
		source_branch: str,
		target_branch: str,
		state: str
		self, id: str, iid: str, source_branch: str, target_branch: str, state: str
		) -> None:
		self.id: str = id
		self.iid: str = iid
		@@ -53,16 +43,10 @@ class ProjectSourcesCache:
		self.state: str = state

		class Tag:
		def __init__(
		self,
		name: str
		) -> None:
		def __init__(self, name: str) -> None:
		self.name: str = name

		def __init__(
		self,
		project: Project
		) -> None:
		def __init__(self, project: Project) -> None:
		self.project = project
		self.tags: dict[str, ProjectSourcesCache.Tag] \| None = None
		self.branches: dict[str, ProjectSourcesCache.Branch] \| None = None
		@@ -73,7 +57,9 @@ class ProjectSourcesCache:
		# get list of branches, with protected and default statuses
		self.branches = dict()
		for branch in self.project.branches.list(iterator=True):
		self.branches[branch.name] = ProjectSourcesCache.Branch(name=branch.name, protected=branch.protected, default=branch.default)
		self.branches[branch.name] = ProjectSourcesCache.Branch(
		name=branch.name, protected=branch.protected, default=branch.default
		)

		if ref in self.branches:
		return self.branches[ref]
		@@ -100,13 +86,17 @@ class ProjectSourcesCache:
		# check if MR is already in the cache
		if merge_request_iid not in self.merge_requests:
		# get from API
		merge_request_from_api = self.project.mergerequests.get(merge_request_iid, lazy=False)
		merge_request_from_api = self.project.mergerequests.get(
		merge_request_iid, lazy=False
		)
		# store MR in cache
		self.merge_requests[merge_request_iid] = ProjectSourcesCache.MergeRequest(id=merge_request_from_api.id,
		self.merge_requests[merge_request_iid] = ProjectSourcesCache.MergeRequest(
		id=merge_request_from_api.id,
		iid=merge_request_from_api.iid,
		source_branch=merge_request_from_api.source_branch,
		target_branch=merge_request_from_api.target_branch,
		state=merge_request_from_api.state)
		state=merge_request_from_api.state,
		)

		return self.merge_requests[merge_request_iid]

		@@ -118,7 +108,7 @@ class PipelineSource:
		source_type: Optional[PipelineSourceEnum] = None,
		protected: Optional[bool] = None,
		default: Optional[bool] = None,
		state: Optional[str] = None
		state: Optional[str] = None,
		) -> None:
		self.existing = existing
		self.source_type = source_type
		@@ -129,7 +119,7 @@ class PipelineSource:
		self.state = state

		def __repr__(self) -> str:
		return f'PipelineSource(existing={self.existing}, source_type={self.source_type}, protected={self.protected}, default={self.default}, state={self.state})'
		return f"PipelineSource(existing={self.existing}, source_type={self.source_type}, protected={self.protected}, default={self.default}, state={self.state})"


		def merge_request_iid_from_ref(ref: str) -> str \| None:
		@@ -140,8 +130,8 @@ def merge_request_iid_from_ref(ref: str) -> str \| None:
		:return: IID or None
		"""
		#
		if ref.startswith('refs/merge-requests/') and ref.endswith('/head'):
		iid = re.search(r'refs/merge-requests/(\d+?)/head', ref)
		if ref.startswith("refs/merge-requests/") and ref.endswith("/head"):
		iid = re.search(r"refs/merge-requests/(\d+?)/head", ref)
		if iid is None:
		return None

		@@ -218,9 +208,13 @@ class Butler:
		return self.default_cfg

		def pipeline_url(self, project: Project, pipeline: ProjectPipeline) -> str:
		return f'{self.client.url}/{project.path_with_namespace}/pipelines/{pipeline.id}'
		return (
		f"{self.client.url}/{project.path_with_namespace}/pipelines/{pipeline.id}"
		)

		def get_max_pipelines_for_source(self, pipeline: ProjectPipeline, cfg: ButlerCfg, source: PipelineSource) -> int:
		def get_max_pipelines_for_source(
		self, pipeline: ProjectPipeline, cfg: ButlerCfg, source: PipelineSource
		) -> int:
		"""
		Compute the limit for this pipeline source
		"""
		@@ -231,19 +225,25 @@ class Butler:
		for pattern in cfg.pipelines.keep.per_branch.by_name.keys():
		if fnmatch(pipeline.ref, pattern):
		if self.verbose and self.debug:
		print(f'Pipeline ref {pipeline.ref} matching branch pattern {pattern} => {cfg.pipelines.keep.per_branch.by_name[pattern]}')
		print(
		f"Pipeline ref {pipeline.ref} matching branch pattern {pattern} => {cfg.pipelines.keep.per_branch.by_name[pattern]}"
		)
		return cfg.pipelines.keep.per_branch.by_name[pattern]

		# check if branch is protected
		if source.protected:
		if self.verbose and self.debug:
		print(f'Pipeline ref {pipeline.ref} is protected => {cfg.pipelines.keep.per_branch.protected_branches}')
		print(
		f"Pipeline ref {pipeline.ref} is protected => {cfg.pipelines.keep.per_branch.protected_branches}"
		)
		return cfg.pipelines.keep.per_branch.protected_branches

		# check if branch is the default branch
		if source.default:
		if self.verbose and self.debug:
		print(f'Pipeline ref {pipeline.ref} is default branch => {cfg.pipelines.keep.per_branch.default_branch}')
		print(
		f"Pipeline ref {pipeline.ref} is default branch => {cfg.pipelines.keep.per_branch.default_branch}"
		)
		return cfg.pipelines.keep.per_branch.default_branch

		# return default branch value
		@@ -254,7 +254,9 @@ class Butler:
		for pattern in cfg.pipelines.keep.per_tag.by_name.keys():
		if fnmatch(pipeline.ref, pattern):
		if self.verbose and self.debug:
		print(f'Pipeline ref {pipeline.ref} matching tag pattern {pattern} => {cfg.pipelines.keep.per_tag.by_name[pattern]}')
		print(
		f"Pipeline ref {pipeline.ref} matching tag pattern {pattern} => {cfg.pipelines.keep.per_tag.by_name[pattern]}"
		)
		return cfg.pipelines.keep.per_tag.by_name[pattern]

		# return default value
		@@ -265,7 +267,9 @@ class Butler:
		for state in cfg.pipelines.keep.per_merge_request.by_status.keys():
		if fnmatch(source.state or "", state):
		if self.verbose and self.debug:
		print(f'Pipeline ref {pipeline.ref} matching MR state {state} => {cfg.pipelines.keep.per_merge_request.by_status[state]}')
		print(
		f"Pipeline ref {pipeline.ref} matching MR state {state} => {cfg.pipelines.keep.per_merge_request.by_status[state]}"
		)
		return cfg.pipelines.keep.per_merge_request.by_status[state]

		# return default value
		@@ -273,7 +277,9 @@ class Butler:

		raise ValueError(f'Unexpected pipeline source "{source.source_type}"')

		def cleaning_enabled_for_source(self, cfg: ButlerCfg, source: PipelineSource) -> bool:
		def cleaning_enabled_for_source(
		self, cfg: ButlerCfg, source: PipelineSource
		) -> bool:
		match source.source_type:
		case PipelineSourceEnum.BRANCH:
		return cfg.pipelines.keep.per_branch.enabled
		@@ -312,8 +318,8 @@ class Butler:

		# check if access level is high enough to process project (*_access can be missing or explicitly set to None)
		access_level = max(
		(project.permissions.get('project_access') or {}).get('access_level', 0),
		(project.permissions.get('group_access') or {}).get('access_level', 0)
		(project.permissions.get("project_access") or {}).get("access_level", 0),
		(project.permissions.get("group_access") or {}).get("access_level", 0),
		)

		if access_level < 30:
		@@ -324,18 +330,21 @@ class Butler:

		return True

		def find_pipeline_source_status(self, project: Project, pipeline: ProjectPipeline, caches: ProjectSourcesCache) -> PipelineSource \| None:
		def find_pipeline_source_status(
		self, project: Project, pipeline: ProjectPipeline, caches: ProjectSourcesCache
		) -> PipelineSource \| None:
		"""
		Check if the source of the pipeline still exists
		"""
		if pipeline.source in ['merge_request_event']:
		if pipeline.source in ["merge_request_event"]:
		merge_request_iid = merge_request_iid_from_ref(pipeline.ref)

		# if MR id cannot be computed (and user allowed processing after an error), skip this pipeline
		if merge_request_iid is None:
		print(
		f'Pipeline {AnsiColors.BLUE}{pipeline.id}{AnsiColors.RESET} was triggered by a merge request event but \'{AnsiColors.BLUE}{pipeline.ref}{AnsiColors.RESET}\' is not a merge request reference. '
		f'Consider deleting this pipeline manually : {self.pipeline_url(project, pipeline)}')
		f"Pipeline {AnsiColors.BLUE}{pipeline.id}{AnsiColors.RESET} was triggered by a merge request event but '{AnsiColors.BLUE}{pipeline.ref}{AnsiColors.RESET}' is not a merge request reference. "
		f"Consider deleting this pipeline manually : {self.pipeline_url(project, pipeline)}"
		)
		return None

		# check if MR is already in the cache and get it from API if needed
		@@ -343,26 +352,47 @@ class Butler:

		# check MR state
		if self.verbose and self.debug:
		considered_as_existing = "no longer existing" if merge_request.state in ['closed', 'merged'] else "existing"
		print(f'Pipeline {pipeline.id} {pipeline.ref} is linked to MR {merge_request.iid} ({merge_request.state} state, considered as {considered_as_existing})')
		considered_as_existing = (
		"no longer existing"
		if merge_request.state in ["closed", "merged"]
		else "existing"
		)
		print(
		f"Pipeline {pipeline.id} {pipeline.ref} is linked to MR {merge_request.iid} ({merge_request.state} state, considered as {considered_as_existing})"
		)

		# MR considering as existing if state is not closed/merged
		return PipelineSource(existing=merge_request.state not in ['closed', 'merged'], source_type=PipelineSourceEnum.MERGE_REQUEST, state=merge_request.state)
		return PipelineSource(
		existing=merge_request.state not in ["closed", "merged"],
		source_type=PipelineSourceEnum.MERGE_REQUEST,
		state=merge_request.state,
		)
		else:
		# pipeline from a branch ?
		if branch := caches.from_branches(pipeline.ref):
		if self.verbose and self.debug:
		print(f'Pipeline {pipeline.id} {pipeline.ref} is linked to branch {pipeline.ref} : {branch}')
		return PipelineSource(existing=True, source_type=PipelineSourceEnum.BRANCH, default=branch.default, protected=branch.protected)
		print(
		f"Pipeline {pipeline.id} {pipeline.ref} is linked to branch {pipeline.ref} : {branch}"
		)
		return PipelineSource(
		existing=True,
		source_type=PipelineSourceEnum.BRANCH,
		default=branch.default,
		protected=branch.protected,
		)

		# pipeline from a tag ?
		if tag := caches.from_tags(pipeline.ref):
		if self.verbose and self.debug:
		print(f'Pipeline {pipeline.id} {pipeline.ref} is linked to tag {pipeline.ref} : {tag}')
		print(
		f"Pipeline {pipeline.id} {pipeline.ref} is linked to tag {pipeline.ref} : {tag}"
		)
		return PipelineSource(existing=True, source_type=PipelineSourceEnum.TAG)

		if self.verbose and self.debug:
		print(f'Pipeline {pipeline.id} {pipeline.ref} ({pipeline.source}) is not linked to any branch or tag')
		print(
		f"Pipeline {pipeline.id} {pipeline.ref} ({pipeline.source}) is not linked to any branch or tag"
		)

		return PipelineSource(existing=False)

		@@ -418,13 +448,20 @@ class Butler:
		:param project:
		:return: a ProjectPipeline iterator
		"""
		iterator = project.pipelines.list(iterator=True, sort="asc", order_by="updated_at")
		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):
		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
		@@ -443,10 +480,12 @@ class Butler:

		cfg = self.load_cfg(project)
		if self.verbose and self.debug:
		print(f'loaded cfg for project {project.path_with_namespace} = {cfg}')
		print(f"loaded cfg for project {project.path_with_namespace} = {cfg}")

		# generate cutoff datetime, from current local day (must be offset-aware)
		updated_limit = datetime.now().astimezone().replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=cfg.pipelines.delete_older_than)
		updated_limit = datetime.now().astimezone().replace(
		hour=0, minute=0, second=0, microsecond=0
		) - timedelta(days=cfg.pipelines.delete_older_than)

		# count deletion attempts, and actual deletions
		pipelines_deletions_attempts = 0
		@@ -461,45 +500,62 @@ class Butler:
		# iterate over project pipelines (starting with oldest)
		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.pipeline_deletion_limit and (
		pipelines_deletions_attempts >= self.pipeline_deletion_limit
		):
		if self.verbose:
		print('Pipeline deletion limit reached, stopping processing for this project.')
		print(
		"Pipeline deletion limit reached, stopping processing for this project."
		)
		break

		# if the pipeline has been updated after the cutoff date: keep and break loop
		if datetime.fromisoformat(pipeline.updated_at) >= updated_limit:
		if self.verbose and self.debug:
		print(
		f'Pipeline {pipeline.id} age is under limit ({datetime.fromisoformat(pipeline.updated_at)} >= {updated_limit}), stopping')
		f"Pipeline {pipeline.id} age is under limit ({datetime.fromisoformat(pipeline.updated_at)} >= {updated_limit}), stopping"
		)
		break

		# check pipeline source status
		pipeline_source = self.find_pipeline_source_status(project, pipeline, caches)
		pipeline_source = self.find_pipeline_source_status(
		project, pipeline, caches
		)

		# skip the pipeline if the source cannot be found
		if pipeline_source is None:
		print(f'Unable to find pipeline {AnsiColors.BLUE}{pipeline.id}{AnsiColors.RESET} source, skipping')
		print(
		f"Unable to find pipeline {AnsiColors.BLUE}{pipeline.id}{AnsiColors.RESET} source, skipping"
		)
		continue

		# skip the pipeline if cleaning source type is not enabled
		if pipeline_source.existing and not self.cleaning_enabled_for_source(cfg, pipeline_source):
		if pipeline_source.existing and not self.cleaning_enabled_for_source(
		cfg, pipeline_source
		):
		continue

		# if source no longer exists, delete pipeline immediately
		if not pipeline_source.existing:
		if self.verbose and self.debug:
		print(f'Pipeline {pipeline.id} source ({pipeline_source}) no longer exists -> delete')
		print(
		f"Pipeline {pipeline.id} source ({pipeline_source}) no longer exists -> delete"
		)
		if self.delete_pipeline(project, pipeline):
		pipelines_deletions_count += 1
		pipelines_deletions_attempts += 1
		continue

		# if source still exists, store the pipeline
		max_pipelines_for_source = self.get_max_pipelines_for_source(pipeline, cfg, pipeline_source)
		max_pipelines_for_source = self.get_max_pipelines_for_source(
		pipeline, cfg, pipeline_source
		)

		if max_pipelines_for_source == 0:
		if self.verbose and self.debug:
		print(f'Pipeline {pipeline.id} source ({pipeline_source}) configured not to keep any -> delete')
		print(
		f"Pipeline {pipeline.id} source ({pipeline_source}) configured not to keep any -> delete"
		)
		if self.delete_pipeline(project, pipeline):
		pipelines_deletions_count += 1
		pipelines_deletions_attempts += 1
		@@ -514,7 +570,9 @@ class Butler:
		while len(kept_pipelines_for_ref) > max_pipelines_for_source:
		pipeline = kept_pipelines_for_ref.pop()
		if self.verbose and self.debug:
		print(f'Pipeline list ({len(kept_pipelines_for_ref)+1}) over limit ({max_pipelines_for_source}) -> delete pipeline')
		print(
		f"Pipeline list ({len(kept_pipelines_for_ref) + 1}) over limit ({max_pipelines_for_source}) -> delete pipeline"
		)
		if self.delete_pipeline(project, pipeline):
		pipelines_deletions_count += 1
		pipelines_deletions_attempts += 1
		@@ -525,9 +583,13 @@ class Butler:

		# display project statistics
		if self.dry_run and pipelines_deletions_attempts > 0:
		print(f" ==> number of pipelines which would have been deleted: {AnsiColors.HPURPLE}{pipelines_deletions_attempts}{AnsiColors.RESET}")
		print(
		f" ==> number of pipelines which would have been deleted: {AnsiColors.HPURPLE}{pipelines_deletions_attempts}{AnsiColors.RESET}"
		)
		elif not self.dry_run and pipelines_deletions_count > 0:
		print(f" ==> number of pipelines which have been deleted: {AnsiColors.HPURPLE}{pipelines_deletions_count}{AnsiColors.RESET}")
		print(
		f" ==> number of pipelines which have been deleted: {AnsiColors.HPURPLE}{pipelines_deletions_count}{AnsiColors.RESET}"
		)

		# Cleanup recursively a GitLab group
		# $1: group to be cleaned
		@@ -568,7 +630,7 @@ class Butler:
		print(
		f"🏢 Group {AnsiColors.BLUE}{subgroup.full_path}{AnsiColors.RESET} matches excludes: {AnsiColors.HGRAY}skip{AnsiColors.RESET}"
		)
		elif (subgroup_id := subgroup.get_id()):
		elif subgroup_id := subgroup.get_id():
		self.clean_group(self.client.groups.get(subgroup_id))
		else:
		print("- skipping sub groups...")

gitlab_butler/butlercfg.py

+9 −2

Original line number	Diff line number	Diff line
		@@ -12,6 +12,7 @@ class ButlerCfg(BaseModel):

		class PerBranchCfg(BaseModel):
		"""Per branch configuration."""

		enabled: bool = True
		default_branch: int = Field(alias="default-branch", default=10)
		protected_branches: int = Field(alias="protected-branches", default=10)
		@@ -28,6 +29,7 @@ class ButlerCfg(BaseModel):

		class PerTagCfg(BaseModel):
		"""Per branch configuration."""

		enabled: bool = True
		any_tag: int = Field(alias="any-tag", default=5)
		by_name: dict[str, int] = Field(
		@@ -40,6 +42,7 @@ class ButlerCfg(BaseModel):

		class PerMergeRequestCfg(BaseModel):
		"""Per merge-request configuration."""

		enabled: bool = True
		any_mr: int = Field(alias="any-mr", default=1)
		by_status: dict[str, int] = Field(
		@@ -52,9 +55,13 @@ class ButlerCfg(BaseModel):
		},
		)

		per_branch: PerBranchCfg = Field(alias="per-branch", default_factory=PerBranchCfg)
		per_branch: PerBranchCfg = Field(
		alias="per-branch", default_factory=PerBranchCfg
		)
		per_tag: PerTagCfg = Field(alias="per-tag", default_factory=PerTagCfg)
		per_merge_request: PerMergeRequestCfg = Field(alias="per-merge-request", default_factory=PerMergeRequestCfg)
		per_merge_request: PerMergeRequestCfg = Field(
		alias="per-merge-request", default_factory=PerMergeRequestCfg
		)

		enabled: bool = True
		keep: KeepCfg = Field(default_factory=KeepCfg)

gitlab_butler/main.py

+12 −10

Original line number	Diff line number	Diff line
		@@ -157,15 +157,17 @@ def run() -> None:
		# parse command and args
		args = parser.parse_args()

		assert args.group_path, "GitLab root group path is required (use --group-path CLI option or GROUP_PATH variable)"
		assert (
		args.token
		), "GitLab token is required (use --token CLI option or GITLAB_TOKEN variable)"
		assert (
		args.pipeline_deletion_limit >= 0
		), "Pipeline deletion limit must be greater or equal than zero"
		assert args.group_path, (
		"GitLab root group path is required (use --group-path CLI option or GROUP_PATH variable)"
		)
		assert args.token, (
		"GitLab token is required (use --token CLI option or GITLAB_TOKEN variable)"
		)
		assert args.pipeline_deletion_limit >= 0, (
		"Pipeline deletion limit must be greater or equal than zero"
		)

		print(f'GitLab-Butler {version('gitlab-butler')}')
		print(f"GitLab-Butler {version('gitlab-butler')}")
		print(
		f"- GitLab API url (--api-url) : {AnsiColors.CYAN}{args.api_url}{AnsiColors.RESET}"
		)
		@@ -264,7 +266,7 @@ def run() -> None:
		f" {AnsiColors.HPURPLE}{butler.pipelines_deletions_attempts}{AnsiColors.RESET} pipelines would be deleted;"
		f" {AnsiColors.RED}{len(butler.warnings)}{AnsiColors.RESET} warnings;"
		f" {AnsiColors.RED}{len(butler.errors)}{AnsiColors.RESET} errors"
		f"{" (see logs)" if butler.warnings or butler.errors else ""}"
		f"{' (see logs)' if butler.warnings or butler.errors else ''}"
		)
		else:
		print(
		@@ -272,7 +274,7 @@ def run() -> None:
		f" {AnsiColors.HPURPLE}{butler.pipelines_count}{AnsiColors.RESET} pipelines deleted;"
		f" {AnsiColors.RED}{len(butler.warnings)}{AnsiColors.RESET} warnings;"
		f" {AnsiColors.RED}{len(butler.errors)}{AnsiColors.RESET} errors"
		f"{" (see logs)" if butler.warnings or butler.errors else ""}"
		f"{' (see logs)' if butler.warnings or butler.errors else ''}"
		)

		if len(butler.errors) > 0:

tests/test_basic_functions.py

+75 −42

File changed.

Preview size limit exceeded, changes collapsed.

tests/test_butlercfg.py

+118 −45

File changed.

Preview size limit exceeded, changes collapsed.