feature: implements repo deletion mechanism

rhodecode/model/repo.py

@@ -23,6 +23,8 @@ import time
 import logging
 import traceback
 import datetime
+from enum import StrEnum, auto
+from typing import Optional
 
 from sqlalchemy.orm import aliased
 from zope.cachedescriptors.property import Lazy as LazyProperty
@@ -76,6 +78,15 @@ log = logging.getLogger(__name__)
 class RepoModel(BaseModel):
     cls = Repository
 
+    class ForkAction(StrEnum):
+        DELETE = auto()
+        DETACH = auto()
+        UNKNOWN = auto()
+
+        @classmethod
+        def _missing_(cls, value):
+            return cls(cls.UNKNOWN)
+
     def _get_user_group(self, users_group):
         return self._get_instance(UserGroup, users_group, callback=UserGroup.get_by_group_name)
 
@@ -749,17 +760,20 @@ class RepoModel(BaseModel):
                 raise
 
     def delete(
-        self, repo, forks=None, pull_requests=None, artifacts=None, fs_remove=True, cur_user=None, call_events=True
-    ):
+        self,
+        repo: Repository,
+        forks: ForkAction = None,  # left for backward compatibility, do not use it
+        fs_remove: bool = True,
+        cur_user: User = None,
+        call_events: bool = True,
+    ) -> bool:
         """
         Delete given repository, forks parameter defines what do do with
         attached forks. Throws AttachedForksError if deleted repo has attached
         forks
 
         :param repo:
-        :param forks: str 'delete' or 'detach'
-        :param pull_requests: str 'delete' or None
-        :param artifacts: str 'delete' or None
+        :param forks: ForkAction 'delete' or 'detach', NOTE: left for backward compatibility, do not use it
         :param fs_remove: remove(archive) repo from filesystem
         """
         if not cur_user:
@@ -768,27 +782,21 @@ class RepoModel(BaseModel):
         if not repo:
             return False
 
-        if not forks:  # left for backward compatibility
+        if forks is None or forks is self.ForkAction.UNKNOWN:
             is_private = repo.private
-            forks = "delete" if is_private else "detach"
+            forks = self.ForkAction.DELETE if is_private else self.ForkAction.DETACH
 
-        if forks == "detach":
+        if forks is self.ForkAction.DETACH:
             for r in repo.forks:
                 r.fork = None
                 self.sa.add(r)
-        elif forks == "delete":
+        elif forks is self.ForkAction.DELETE:
             for r in repo.forks:
-                self.delete(r, forks="delete")
-        elif [f for f in repo.forks]:
+                self.delete(r, forks=self.ForkAction.DELETE)
+        elif repo.forks.first() is not None:
             raise AttachedForksError()
 
-        # check for pull requests
-        pr_sources = repo.pull_requests_source
-        pr_targets = repo.pull_requests_target
-        artifacts = repo.artifacts
-        # FIXME: continue here
-        for pr in itertools.chain(pr_sources, pr_targets, artifacts):
-            self.sa.delete(pr)
+        self._delete_repo_related_assets(repo)
 
         old_repo_dict = repo.get_dict()
         if call_events:
@@ -815,6 +823,14 @@ class RepoModel(BaseModel):
 
         return True
 
+    def _delete_repo_related_assets(self, repo):
+        pr_sources = repo.pull_requests_source
+        pr_targets = repo.pull_requests_target
+        artifacts = repo.artifacts
+        logs = repo.logs
+        for obj_to_remove in itertools.chain(pr_sources, pr_targets, artifacts, logs):
+            self.sa.delete(obj_to_remove)
+
     def grant_user_permission(self, repo, user, perm):
         """
         Grant permission for user on given repository, or update existing one