engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feature: implemented deleting user and migrating all user assets new system ghost user

Browse source
This commit is contained in:
ievgenii vdovenko 2025-07-24 16:01:44 +02:00
parent 6184fdeb79
commit 6c5b9aa0de
18 changed files with 5866 additions and 268 deletions
Download patch file Download diff file Expand all files Collapse all files

2
rhodecode/__init__.py
View file

@ -100,7 +100,7 @@ PYRAMID_SETTINGS = {}
EXTENSIONS = {}
__version__ = ".".join((str(each) for each in VERSION[:3]))
__dbversion__ = 116 # defines current db version for migrations
__dbversion__ = 117 # defines current db version for migrations
__license__ = "AGPLv3, and Commercial License"
__author__ = "RhodeCode GmbH"
__url__ = "https://code.rhodecode.com"

4
rhodecode/api/views/user_api.py
View file

@ -145,7 +145,9 @@ def get_users(request, apiuser):
raise JSONRPCForbidden()
result = []
users_list = User.query().order_by(User.username).filter(User.username != User.DEFAULT_USER).all()
users_list = (
User.query().order_by(User.username).filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])).all()
)
for user in users_list:
result.append(user.get_api_data(include_secrets=True))
return result

8
rhodecode/apps/_base/__init__.py
View file

@ -144,7 +144,7 @@ class BaseAppView(object):
if not user_obj:
return
if user_obj.username == User.DEFAULT_USER:
if user_obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
return
now = time.time()
@ -270,7 +270,7 @@ class RepoAppView(BaseAppView):
def _prepare_and_set_clone_url(self, c):
username = ""
if self._rhodecode_user.username != User.DEFAULT_USER:
if self._rhodecode_user.username in [User.DEFAULT_USER, User.GHOST_USER]:
username = self._rhodecode_user.username
_def_clone_uri = c.clone_uri_tmpl
@ -558,9 +558,9 @@ class UserAppView(BaseAppView):
_ = self.request.translate
if not request.db_user_supports_default:
if self.db_user.username == User.DEFAULT_USER:
if self.db_user.username in [User.DEFAULT_USER, User.GHOST_USER]:
h.flash(
_("Editing user `{}` is disabled.".format(User.DEFAULT_USER)),
_("Editing user `{}` is disabled.".format(self.db_user.username)),
category="warning",
)
raise HTTPFound(h.route_path("users"))

99
rhodecode/apps/admin/views/users.py
View file

@ -104,14 +104,13 @@ class AdminUsersView(BaseAppView, DataGridAppView):
def user_actions(user_id, username):
return _render("user_actions", user_id, username)
users_data_total_count = User.query().filter(User.username != User.DEFAULT_USER).count()
excluded_users = User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])
users_data_total_count = User.query().filter(excluded_users).count()
users_data_total_inactive_count = (
User.query().filter(User.username != User.DEFAULT_USER).filter(User.active != true()).count()
)
users_data_total_inactive_count = User.query().filter(excluded_users).filter(User.active != true()).count()
# json generate
base_q = User.query().filter(User.username != User.DEFAULT_USER)
base_q = User.query().filter(excluded_users)
base_inactive_q = base_q.filter(User.active != true())
if search_q:
@ -367,90 +366,54 @@ class UsersView(UserAppView):
_pull_requests = len(c.user.user_pull_requests)
_artifacts = len(c.user.artifacts)
handle_repos = None
handle_repo_groups = None
handle_user_groups = None
handle_pull_requests = None
handle_artifacts = None
# calls for flash of handle based on handle case detach or delete
def set_handle_flash_repos():
handle = handle_repos
if handle == "detach":
def set_flash_repos_if_detached():
if detach_repos:
h.flash(_("Detached %s repositories") % _repos, category="success")
elif handle == "delete":
h.flash(_("Deleted %s repositories") % _repos, category="success")
def set_handle_flash_repo_groups():
handle = handle_repo_groups
if handle == "detach":
def set_flash_repo_groups_if_detached():
if detach_repo_groups:
h.flash(_("Detached %s repository groups") % _repo_groups, category="success")
elif handle == "delete":
h.flash(_("Deleted %s repository groups") % _repo_groups, category="success")
def set_handle_flash_user_groups():
handle = handle_user_groups
if handle == "detach":
def set_flash_user_groups_if_detached():
if detach_user_groups:
h.flash(_("Detached %s user groups") % _user_groups, category="success")
elif handle == "delete":
h.flash(_("Deleted %s user groups") % _user_groups, category="success")
def set_handle_flash_pull_requests():
handle = handle_pull_requests
if handle == "detach":
def set_flash_pull_requests_if_detached():
if detach_pull_requests:
h.flash(_("Detached %s pull requests") % _pull_requests, category="success")
elif handle == "delete":
h.flash(_("Deleted %s pull requests") % _pull_requests, category="success")
def set_handle_flash_artifacts():
handle = handle_artifacts
if handle == "detach":
def set_flash_artifacts_if_detached():
if detach_artifacts:
h.flash(_("Detached %s artifacts") % _artifacts, category="success")
elif handle == "delete":
h.flash(_("Deleted %s artifacts") % _artifacts, category="success")
handle_user = User.get_first_super_admin()
handle_user_id = safe_int(self.request.POST.get("detach_user_id"))
if handle_user_id:
# NOTE(marcink): we get new owner for objects...
handle_user = User.get_or_404(handle_user_id)
if _repos and self.request.POST.get("user_repos"):
handle_repos = self.request.POST["user_repos"]
if _repo_groups and self.request.POST.get("user_repo_groups"):
handle_repo_groups = self.request.POST["user_repo_groups"]
if _user_groups and self.request.POST.get("user_user_groups"):
handle_user_groups = self.request.POST["user_user_groups"]
if _pull_requests:
handle_pull_requests = "detach"
if _artifacts and self.request.POST.get("user_artifacts"):
handle_artifacts = self.request.POST["user_artifacts"]
detach_repos = _repos > 0
detach_repo_groups = _repo_groups > 0
detach_user_groups = _user_groups > 0
detach_pull_requests = _pull_requests > 0
detach_artifacts = _artifacts > 0
old_values = c.user.get_api_data()
try:
ghost_user = User.get_ghost_user()
UserModel().delete(
c.user,
handle_repos=handle_repos,
handle_repo_groups=handle_repo_groups,
handle_user_groups=handle_user_groups,
handle_pull_requests=handle_pull_requests,
handle_artifacts=handle_artifacts,
handle_new_owner=handle_user,
detach_repos=detach_repos,
detach_repo_groups=detach_repo_groups,
detach_user_groups=detach_user_groups,
detach_pull_requests=detach_pull_requests,
detach_artifacts=detach_artifacts,
new_owner=ghost_user,
)
audit_logger.store_web("user.delete", action_data={"old_data": old_values}, user=c.rhodecode_user)
Session().commit()
set_handle_flash_repos()
set_handle_flash_repo_groups()
set_handle_flash_user_groups()
set_handle_flash_pull_requests()
set_handle_flash_artifacts()
set_flash_repos_if_detached()
set_flash_repo_groups_if_detached()
set_flash_user_groups_if_detached()
set_flash_pull_requests_if_detached()
set_flash_artifacts_if_detached()
username = h.escape(old_values["username"])
h.flash(_("Successfully deleted user `{}`").format(username), category="success")
except (

8
rhodecode/apps/home/views.py
View file

@ -78,12 +78,12 @@ class HomeView(BaseAppView, DataGridAppView):
_users = UserModel().get_users(name_contains=query, only_active=active)
def maybe_skip_default_user(usr):
if skip_default_user and usr["username"] == UserModel.cls.DEFAULT_USER:
def maybe_skip_user(usr):
if skip_default_user and usr["username"] in [UserModel.cls.DEFAULT_USER, UserModel.cls.GHOST_USER]:
return False
return True
_users = list(filter(maybe_skip_default_user, _users))
_users = list(filter(maybe_skip_user, _users))
if include_groups:
# extend with user groups
@ -239,7 +239,7 @@ class HomeView(BaseAppView, DataGridAppView):
User.query()
.order_by(func.length(User.username))
.order_by(User.username)
.filter(User.username != User.DEFAULT_USER)
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER]))
)
if name_contains:

2
rhodecode/apps/user_group_profile/views.py
View file

@ -46,5 +46,5 @@ class UserGroupProfileView(BaseAppView):
raise HTTPNotFound()
group_members_obj = sorted((x.user for x in c.user_group.members), key=lambda u: u.username.lower())
c.group_members = group_members_obj
c.anonymous = self._rhodecode_user.username == User.DEFAULT_USER
c.anonymous = self._rhodecode_user.username in [User.DEFAULT_USER, User.GHOST_USER]
return self._get_template_context(c)

2
rhodecode/apps/user_profile/views.py
View file

@ -40,7 +40,7 @@ class UserProfileView(BaseAppView):
username = self.request.matchdict.get("username")
c.user = UserModel().get_by_username(username)
if not c.user or c.user.username == User.DEFAULT_USER:
if not c.user or c.user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise HTTPNotFound()
return self._get_template_context(c)

4
rhodecode/authentication/plugins/auth_rhodecode.py
View file

@ -138,6 +138,10 @@ class RhodeCodeAuthPlugin(RhodeCodeAuthPluginBase):
log.warning("userobj:%s is not super-admin and auth restriction is set to %s", userobj, auth_restriction)
return None
if userobj.username == User.GHOST_USER:
log.warning("userobj:%s is not allowed to login", userobj)
return None
user_attrs = {
"username": userobj.username,
"firstname": userobj.firstname,

8
rhodecode/lib/auth.py
View file

@ -1295,6 +1295,10 @@ class AuthUser(object):
def is_default(self):
return self.username == User.DEFAULT_USER
@property
def is_ghost(self):
return self.username == User.GHOST_USER
@property
def is_admin(self):
return self.admin
@ -1878,7 +1882,7 @@ class NotAnonymous(object):
_ = request.translate
log.debug("Checking if user is not anonymous @%s", cls)
anonymous = self.user.username == User.DEFAULT_USER
anonymous = self.user.username in [User.DEFAULT_USER, User.GHOST_USER]
if anonymous:
came_from = get_came_from(request)
@ -1918,7 +1922,7 @@ class PermsDecorator(object):
else:
log.debug("Permission denied for %s %s", cls, _user)
anonymous = _user.username == User.DEFAULT_USER
anonymous = _user.username in [User.DEFAULT_USER, User.GHOST_USER]
if anonymous:
import rhodecode.lib.helpers as h

5670
rhodecode/lib/dbmigrate/schema/db_4_20_0_1.py Normal file
View file

File diff suppressed because it is too large Load diff

46
rhodecode/lib/dbmigrate/versions/117_version_5_7_0.py Normal file
View file

@ -0,0 +1,46 @@
import json
import logging
from psycopg2.errorcodes import DUPLICATE_COLUMN
from sqlalchemy import *
from sqlalchemy.engine import reflection
from alembic.migration import MigrationContext
from alembic.operations import Operations
from rhodecode.lib.dbmigrate.versions import _reset_base
from rhodecode.lib.jsonalchemy import MutationObj, JsonType
from rhodecode.model import meta, init_model_encryption
from rhodecode.model.db import User
def upgrade(migrate_engine):
"""
Upgrade operations go here.
Don't create your own engine; bind migrate_engine to your metadata
"""
_reset_base(migrate_engine)
from rhodecode.lib.dbmigrate.schema import db_4_20_0_1 as db
init_model_encryption(db)
UserModel: User = db.User
ghost_user = UserModel()
ghost_user.name = UserModel.GHOST_USER.capitalize()
ghost_user.username = UserModel.GHOST_USER
ghost_user.lastname = "User"
ghost_user.strict_creation_check = False
ghost_user.email = UserModel.GHOST_USER_EMAIL
ghost_user.active = False
ghost_user.description = "System user to own assets of deleted users (does not consume license slot)"
ghost_user.is_new_user = False
session = meta.Session()
session.add(ghost_user)
session.commit()
def downgrade(migrate_engine):
pass

2
rhodecode/lib/helpers.py
View file

@ -1035,7 +1035,7 @@ def link_to_user(author, length=0, **kwargs):
if length:
display_person = shorter(display_person, length)
if user and user.username != user.DEFAULT_USER:
if user and user.username not in [user.DEFAULT_USER, user.GHOST_USER]:
return link_to(escape(display_person), route_path("user_profile", username=user.username), **kwargs)
else:
return escape(display_person)

2
rhodecode/lib/middleware/simplevcs.py
View file

@ -484,7 +484,7 @@ class SimpleVCS(object):
auth_user = user_obj.AuthUser()
if (
user_obj
and user_obj.username != User.DEFAULT_USER
and user_obj.username not in [User.DEFAULT_USER, User.GHOST_USER]
and user_obj.user_data.get("force_password_change")
):
reason = "password change required"

7
rhodecode/lib/utils.py
View file

@ -536,9 +536,6 @@ def get_rhodecode_repo_store_path():
Returns the base path. The base path is the filesystem path which points
to the repository store.
"""
import rhodecode
return rhodecode.CONFIG["repo_store.path"]
@ -756,7 +753,6 @@ def deep_reload_package(package_name):
def load_rcextensions(root_path):
import rhodecode
from rhodecode.config import conf
path = os.path.join(root_path)
@ -785,7 +781,6 @@ def get_custom_lexer(extension):
returns a custom lexer if it is defined in rcextensions module, or None
if there's no custom lexer defined
"""
import rhodecode
from pygments import lexers
# custom override made by RhodeCode
@ -881,7 +876,7 @@ def create_test_repositories(test_path, config):
def password_changed(auth_user, session):
# Never report password change in case of default user or anonymous user.
if auth_user.username == User.DEFAULT_USER or auth_user.user_id is None:
if auth_user.username in [User.DEFAULT_USER, User.GHOST_USER] or auth_user.user_id is None:
return False
password_hash = md5(safe_bytes(auth_user.password)) if auth_user.password else None

17
rhodecode/model/db.py
View file

@ -134,11 +134,11 @@ PERMISSION_TYPE_SORT = {
def display_user_sort(obj):
"""
Sort function used to sort permissions in .permissions() function of
Repository, RepoGroup, UserGroup. Also it put the default user in front
Repository, RepoGroup, UserGroup. Also it put the default and ghost users in front
of all other resources
"""
if obj.username == User.DEFAULT_USER:
if obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
return "#####"
prefix = PERMISSION_TYPE_SORT.get(obj.permission.split(".")[-1], "")
extra_sort_num = "1" # default
@ -617,6 +617,8 @@ class User(Base, BaseModel):
DEFAULT_USER = "default"
DEFAULT_USER_EMAIL = "anonymous@rhodecode.org"
GHOST_USER = "ghost"
GHOST_USER_EMAIL = "ghost@rhodecode.org"
DEFAULT_GRAVATAR_URL = "https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}"
RECOVERY_CODES_COUNT = 10
@ -1259,6 +1261,15 @@ class User(Base, BaseModel):
raise Exception("FATAL: Missing administrative account!")
return user
@classmethod
def get_ghost_user(cls):
stmt = cls.select().where(User.username == User.GHOST_USER)
user = cls.scalars(stmt).first()
if user is None:
raise Exception("FATAL: Missing ghost account!")
return user
@classmethod
def get_all_super_admins(cls, only_active=False):
"""
@ -3037,7 +3048,7 @@ class RepoGroup(Base, BaseModel):
@classmethod
def get_user_personal_repo_group(cls, user_id):
user = User.get(user_id)
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
return None
return cls.query().filter(cls.personal == true()).filter(cls.user == user).order_by(cls.group_id.asc()).first()

4
rhodecode/model/repo.py
View file

@ -664,6 +664,10 @@ class RepoModel(BaseModel):
member_id = int(member_id)
if member_type == "user":
member_name = User.get(member_id).username
if member_name == User.GHOST_USER:
raise ValueError("Modify user permissions for ghost account is not allowed.")
if member_name == User.DEFAULT_USER:
# NOTE(dan): detect if we changed permissions for default user
perm_obj = (

10
rhodecode/model/repo_group.py
View file

@ -359,7 +359,7 @@ class RepoGroupModel(BaseModel):
elif isinstance(_obj, Repository):
# private repos will not allow to change the default
# permissions using recursive mode
if _obj.private and _user_obj.username == User.DEFAULT_USER:
if _obj.private and _user_obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
log.debug("Skipping private repo %s for user %s", _obj, _user_obj)
return
@ -381,7 +381,7 @@ class RepoGroupModel(BaseModel):
elif isinstance(_obj, Repository):
# private repos will not allow to change the default
# permissions using recursive mode, also there's no revocation fo default user, just update
if _user_obj.username == User.DEFAULT_USER:
if _user_obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
log.debug("Skipping private repo %s for user %s", _obj, _user_obj)
return
RepoModel().revoke_user_permission(repo=_obj, user=_user_obj)
@ -425,7 +425,11 @@ class RepoGroupModel(BaseModel):
if member_type == "user":
member_obj = User.get(member_id)
member_name = member_obj.username
if isinstance(obj, RepoGroup) and obj == repo_group and member_name == User.DEFAULT_USER:
is_repo_group = isinstance(obj, RepoGroup) and obj == repo_group
if is_repo_group and member_name == User.GHOST_USER:
raise ValueError("Modify user permissions for ghost account is not allowed.")
if is_repo_group and member_name == User.DEFAULT_USER:
# NOTE(dan): detect if we changed permissions for default user
perm_obj = (
self.sa.query(UserRepoGroupToPerm)

239
rhodecode/model/user.py
View file

@ -135,7 +135,11 @@ class UserModel(BaseModel):
return User.get_by_auth_token(auth_token, cache)
def get_active_user_count(self, cache=False):
qry = User.query().filter(User.active == true()).filter(User.username != User.DEFAULT_USER)
qry = (
User.query()
.filter(User.active == true())
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER]))
)
if cache:
qry = qry.options(FromCache("sql_cache_short", "get_active_users"))
return qry.count()
@ -169,10 +173,8 @@ class UserModel(BaseModel):
raise
def update_user(self, user, skip_attrs=None, **kwargs):
from rhodecode.lib.auth import get_crypt_password
user = self._get_user(user)
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise DefaultUserException(
"You can't edit this user (`%(username)s`) since it's "
"crucial for entire application" % {"username": user.username}
@ -356,7 +358,7 @@ class UserModel(BaseModel):
edit = True
# we're not allowed to edit default user
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise DefaultUserException(
"You can't edit this user (`%(username)s`) since it's "
"crucial for entire application" % {"username": user.username}
@ -459,131 +461,60 @@ class UserModel(BaseModel):
log.error(traceback.format_exc())
raise
def _handle_user_repos(self, username, repositories, handle_user, handle_mode=None):
left_overs = True
def _detach_user_repos(self, username, repositories, handle_user):
for obj in repositories:
obj.user = handle_user
# set description we know why ghost now owns
# additional repositories that were orphaned !
obj.description += " \n::detached repository from deleted user: %s" % (username,)
self.sa.add(obj)
from rhodecode.model.repo import RepoModel
def _detach_user_repo_groups(self, username, repository_groups, handle_user):
for r in repository_groups:
r.user = handle_user
# set description we know why ghost now owns
# additional repositories that were orphaned !
r.group_description += " \n::detached repository group from deleted user: %s" % (username,)
r.personal = False
self.sa.add(r)
if handle_mode == "detach":
for obj in repositories:
obj.user = handle_user
# set description we know why we super admin now owns
# additional repositories that were orphaned !
obj.description += " \n::detached repository from deleted user: %s" % (username,)
self.sa.add(obj)
left_overs = False
elif handle_mode == "delete":
for obj in repositories:
RepoModel().delete(obj, forks="detach")
left_overs = False
def _detach_user_user_groups(self, username, user_groups, handle_user):
for r in user_groups:
for user_user_group_to_perm in r.user_user_group_to_perm:
if user_user_group_to_perm.user.username == username:
user_user_group_to_perm.user = handle_user
r.user = handle_user
# set description we know why ghost now owns
# additional repositories that were orphaned !
r.user_group_description += " \n::detached user group from deleted user: %s" % (username,)
self.sa.add(r)
# if nothing is done we have left overs left
return left_overs
def _detach_user_pull_requests(self, username, pull_requests, handle_user):
for pr in pull_requests:
pr.user_id = handle_user.user_id
# set description we know why ghost now owns
# additional repositories that were orphaned !
pr.description += " \n::detached pull requests from deleted user: %s" % (username,)
self.sa.add(pr)
def _handle_user_repo_groups(self, username, repository_groups, handle_user, handle_mode=None):
left_overs = True
from rhodecode.model.repo_group import RepoGroupModel
if handle_mode == "detach":
for r in repository_groups:
r.user = handle_user
# set description we know why we super admin now owns
# additional repositories that were orphaned !
r.group_description += " \n::detached repository group from deleted user: %s" % (username,)
r.personal = False
self.sa.add(r)
left_overs = False
elif handle_mode == "delete":
for r in repository_groups:
RepoGroupModel().delete(r)
left_overs = False
# if nothing is done we have left overs left
return left_overs
def _handle_user_user_groups(self, username, user_groups, handle_user, handle_mode=None):
left_overs = True
from rhodecode.model.user_group import UserGroupModel
if handle_mode == "detach":
for r in user_groups:
for user_user_group_to_perm in r.user_user_group_to_perm:
if user_user_group_to_perm.user.username == username:
user_user_group_to_perm.user = handle_user
r.user = handle_user
# set description we know why we super admin now owns
# additional repositories that were orphaned !
r.user_group_description += " \n::detached user group from deleted user: %s" % (username,)
self.sa.add(r)
left_overs = False
elif handle_mode == "delete":
for r in user_groups:
UserGroupModel().delete(r)
left_overs = False
# if nothing is done we have left overs left
return left_overs
def _handle_user_pull_requests(self, username, pull_requests, handle_user, handle_mode=None):
left_overs = True
from rhodecode.model.pull_request import PullRequestModel
if handle_mode == "detach":
for pr in pull_requests:
pr.user_id = handle_user.user_id
# set description we know why we super admin now owns
# additional repositories that were orphaned !
pr.description += " \n::detached pull requests from deleted user: %s" % (username,)
self.sa.add(pr)
left_overs = False
elif handle_mode == "delete":
for pr in pull_requests:
PullRequestModel().delete(pr)
left_overs = False
# if nothing is done we have leftovers left
return left_overs
def _handle_user_artifacts(self, username, artifacts, handle_user, handle_mode=None):
left_overs = True
if handle_mode == "detach":
for a in artifacts:
a.upload_user = handle_user
# set description we know why we super admin now owns
# additional artifacts that were orphaned !
a.file_description += " \n::detached artifact from deleted user: %s" % (username,)
self.sa.add(a)
left_overs = False
elif handle_mode == "delete":
from rhodecode.apps.file_store import utils as store_utils
request = get_current_request()
f_store = store_utils.get_filestore_backend(request.registry.settings)
for a in artifacts:
file_uid = a.file_uid
f_store.delete(file_uid)
self.sa.delete(a)
left_overs = False
# if nothing is done we have left overs left
return left_overs
def _detach_user_artifacts(self, username, artifacts, handle_user):
for a in artifacts:
a.upload_user = handle_user
# set description we know why ghost now owns
# additional artifacts that were orphaned !
a.file_description += " \n::detached artifact from deleted user: %s" % (username,)
self.sa.add(a)
def delete(
self,
user,
cur_user=None,
handle_repos=None,
handle_repo_groups=None,
handle_user_groups=None,
handle_pull_requests=None,
handle_artifacts=None,
handle_new_owner=None,
detach_repos=False,
detach_repo_groups=False,
detach_user_groups=False,
detach_pull_requests=False,
detach_artifacts=False,
new_owner=None,
):
from rhodecode.lib import hooks_base
@ -593,63 +524,27 @@ class UserModel(BaseModel):
user = self._get_user(user)
try:
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise DefaultUserException("You can't remove this user since it's crucial for entire application")
handle_user = handle_new_owner or self.cls.get_first_super_admin()
log.debug("New detached objects owner %s", handle_user)
left_overs = self._handle_user_repos(user.username, user.repositories, handle_user, handle_repos)
if left_overs and user.repositories:
repos = [x.repo_name for x in user.repositories]
raise UserOwnsReposException(
'user "%(username)s" still owns %(len_repos)s repositories and cannot be '
"removed. Switch owners or remove those repositories:%(list_repos)s"
% {"username": user.username, "len_repos": len(repos), "list_repos": ", ".join(repos)}
)
if detach_repos or detach_repo_groups or detach_user_groups or detach_pull_requests or detach_artifacts:
new_owner = new_owner or self.cls.get_ghost_user()
log.debug("New detached objects owner %s", new_owner)
left_overs = self._handle_user_repo_groups(
user.username, user.repository_groups, handle_user, handle_repo_groups
)
if left_overs and user.repository_groups:
repo_groups = [x.group_name for x in user.repository_groups]
raise UserOwnsRepoGroupsException(
'user "%(username)s" still owns %(len_repo_groups)s repository groups and cannot be '
"removed. Switch owners or remove those repository groups:%(list_repo_groups)s"
% {
"username": user.username,
"len_repo_groups": len(repo_groups),
"list_repo_groups": ", ".join(repo_groups),
}
)
if detach_repos:
self._detach_user_repos(user.username, user.repositories, new_owner)
left_overs = self._handle_user_user_groups(user.username, user.user_groups, handle_user, handle_user_groups)
if left_overs and user.user_groups:
user_groups = [x.users_group_name for x in user.user_groups]
raise UserOwnsUserGroupsException(
'user "%s" still owns %s user groups and cannot be '
"removed. Switch owners or remove those user groups:%s"
% (user.username, len(user_groups), ", ".join(user_groups))
)
if detach_repo_groups:
self._detach_user_repo_groups(user.username, user.repository_groups, new_owner)
left_overs = self._handle_user_pull_requests(
user.username, user.user_pull_requests, handle_user, handle_pull_requests
)
if left_overs and user.user_pull_requests:
pull_requests = [f"!{x.pull_request_id}" for x in user.user_pull_requests]
raise UserOwnsPullRequestsException(
'user "%s" still owns %s pull requests and cannot be '
"removed. Switch owners or remove those pull requests:%s"
% (user.username, len(pull_requests), ", ".join(pull_requests))
)
if detach_user_groups:
self._detach_user_user_groups(user.username, user.user_groups, new_owner)
left_overs = self._handle_user_artifacts(user.username, user.artifacts, handle_user, handle_artifacts)
if left_overs and user.artifacts:
artifacts = [x.file_uid for x in user.artifacts]
raise UserOwnsArtifactsException(
'user "%s" still owns %s artifacts and cannot be '
"removed. Switch owners or remove those artifacts:%s"
% (user.username, len(artifacts), ", ".join(artifacts))
)
if detach_pull_requests:
self._detach_user_pull_requests(user.username, user.user_pull_requests, new_owner)
if detach_artifacts:
self._detach_user_artifacts(user.username, user.artifacts, new_owner)
user_data = user.get_dict() # fetch user data before expire