78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
|
from peewee import JOIN_LEFT_OUTER
|
||
|
|
from cachetools import lru_cache
|
||
|
|
|
||
|
|
from data.database import (Repository, User, Team, TeamMember, RepositoryPermission, TeamRole,
|
||
|
|
Namespace, Visibility, db_for_update)
|
||
|
|
|
||
|
|
|
||
|
|
def get_existing_repository(namespace_name, repository_name, for_update=False):
|
||
|
|
query = (Repository
|
||
|
|
.select(Repository, Namespace)
|
||
|
|
.join(Namespace, on=(Repository.namespace_user == Namespace.id))
|
||
|
|
.where(Namespace.username == namespace_name, Repository.name == repository_name))
|
||
|
|
if for_update:
|
||
|
|
query = db_for_update(query)
|
||
|
|
|
||
|
|
return query.get()
|
||
|
|
|
||
|
|
|
||
|
|
@lru_cache(maxsize=1)
|
||
|
|
def get_public_repo_visibility():
|
||
|
|
return Visibility.get(name='public')
|
||
|
|
|
||
|
|
|
||
|
|
def filter_to_repos_for_user(query, username=None, namespace=None, include_public=True):
|
||
|
|
if not include_public and not username:
|
||
|
|
return Repository.select().where(Repository.id == '-1')
|
||
|
|
|
||
|
|
where_clause = None
|
||
|
|
if username:
|
||
|
|
UserThroughTeam = User.alias()
|
||
|
|
Org = User.alias()
|
||
|
|
AdminTeam = Team.alias()
|
||
|
|
AdminTeamMember = TeamMember.alias()
|
||
|
|
AdminUser = User.alias()
|
||
|
|
|
||
|
|
query = (query
|
||
|
|
.switch(RepositoryPermission)
|
||
|
|
.join(User, JOIN_LEFT_OUTER)
|
||
|
|
.switch(RepositoryPermission)
|
||
|
|
.join(Team, JOIN_LEFT_OUTER)
|
||
|
|
.join(TeamMember, JOIN_LEFT_OUTER)
|
||
|
|
.join(UserThroughTeam, JOIN_LEFT_OUTER, on=(UserThroughTeam.id == TeamMember.user))
|
||
|
|
.switch(Repository)
|
||
|
|
.join(Org, JOIN_LEFT_OUTER, on=(Repository.namespace_user == Org.id))
|
||
|
|
.join(AdminTeam, JOIN_LEFT_OUTER, on=(Org.id == AdminTeam.organization))
|
||
|
|
.join(TeamRole, JOIN_LEFT_OUTER, on=(AdminTeam.role == TeamRole.id))
|
||
|
|
.switch(AdminTeam)
|
||
|
|
.join(AdminTeamMember, JOIN_LEFT_OUTER, on=(AdminTeam.id == AdminTeamMember.team))
|
||
|
|
.join(AdminUser, JOIN_LEFT_OUTER, on=(AdminTeamMember.user == AdminUser.id)))
|
||
|
|
|
||
|
|
where_clause = ((User.username == username) | (UserThroughTeam.username == username) |
|
||
|
|
((AdminUser.username == username) & (TeamRole.name == 'admin')))
|
||
|
|
|
||
|
|
if namespace:
|
||
|
|
where_clause = where_clause & (Namespace.username == namespace)
|
||
|
|
|
||
|
|
# TODO(jschorr, jake): Figure out why the old join on Visibility was so darn slow and
|
||
|
|
# remove this hack.
|
||
|
|
if include_public:
|
||
|
|
new_clause = (Repository.visibility == get_public_repo_visibility())
|
||
|
|
if where_clause:
|
||
|
|
where_clause = where_clause | new_clause
|
||
|
|
else:
|
||
|
|
where_clause = new_clause
|
||
|
|
|
||
|
|
return query.where(where_clause)
|
||
|
|
|
||
|
|
|
||
|
|
def get_user_organizations(username):
|
||
|
|
UserAlias = User.alias()
|
||
|
|
return (User
|
||
|
|
.select()
|
||
|
|
.distinct()
|
||
|
|
.join(Team)
|
||
|
|
.join(TeamMember)
|
||
|
|
.join(UserAlias, on=(UserAlias.id == TeamMember.user))
|
||
|
|
.where(User.organization == True, UserAlias.username == username))
|