vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

refactor(endpoints/api/robot*): adding in database interface

Browse source 
this creates a layer of abstraction so we can move to v2-2 easier

Issue: https://coreosdev.atlassian.net/browse/QUAY-630

- [ ] It works!
- [ ] Comments provide sufficient explanations for the next contributor
- [ ] Tests cover changes and corner cases
- [ ] Follows Quay syntax patterns and format
This commit is contained in:
Charlton Austin 2017-07-25 13:41:20 -04:00
parent c271b1f386
commit 7d5e4dd6d3
3 changed files with 276 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
endpoints/api/robot.py
View file

@ -3,77 +3,24 @@
from endpoints.api import (resource, nickname, ApiResource, log_action, related_user_resource,
require_user_admin, require_scope, path_param, parse_args,
truthy_bool, query_param)
from endpoints.api.robot_models_pre_oci import pre_oci_model as model
from endpoints.exception import Unauthorized
from auth.permissions import AdministerOrganizationPermission, OrganizationMemberPermission
from auth.auth_context import get_authenticated_user
from auth import scopes
from data import model
from data.database import User, Team, Repository, FederatedLogin
from util.names import format_robot_username
from flask import abort
from app import avatar
def robot_view(name, token):
return {
'name': name,
'token': token
}
def permission_view(permission):
return {
'repository': {
'name': permission.repository.name,
'is_public': permission.repository.visibility.name == 'public'
},
'role': permission.role.name
}
def robots_list(prefix, include_permissions=False):
tuples = model.user.list_entity_robot_permission_teams(prefix,
include_permissions=include_permissions)
robots = model.list_entity_robot_permission_teams(prefix, include_permissions=include_permissions)
return {'robots': [robot.to_dict() for robot in robots]}
robots = {}
robot_teams = set()
for robot_tuple in tuples:
robot_name = robot_tuple.get(User.username)
if not robot_name in robots:
robots[robot_name] = {
'name': robot_name,
'token': robot_tuple.get(FederatedLogin.service_ident)
}
if include_permissions:
robots[robot_name].update({
'teams': [],
'repositories': []
})
if include_permissions:
team_name = robot_tuple.get(Team.name)
repository_name = robot_tuple.get(Repository.name)
if team_name is not None:
check_key = robot_name + ':' + team_name
if not check_key in robot_teams:
robot_teams.add(check_key)
robots[robot_name]['teams'].append({
'name': team_name,
'avatar': avatar.get_data(team_name, team_name, 'team')
})
if repository_name is not None:
if not repository_name in robots[robot_name]['repositories']:
robots[robot_name]['repositories'].append(repository_name)
return {'robots': robots.values()}
@resource('/v1/user/robots')
class UserRobotList(ApiResource):
""" Resource for listing user robots. """
@require_user_admin
@nickname('getUserRobots')
@parse_args()
@ -91,29 +38,30 @@ class UserRobotList(ApiResource):
'The short name for the robot, without any user or organization prefix')
class UserRobot(ApiResource):
""" Resource for managing a user's robots. """
@require_user_admin
@nickname('getUserRobot')
def get(self, robot_shortname):
""" Returns the user's robot with the specified name. """
parent = get_authenticated_user()
robot, password = model.user.get_robot(robot_shortname, parent)
return robot_view(robot.username, password)
robot = model.get_user_robot(robot_shortname, parent)
return robot.to_dict()
@require_user_admin
@nickname('createUserRobot')
def put(self, robot_shortname):
""" Create a new user robot with the specified name. """
parent = get_authenticated_user()
robot, password = model.user.create_robot(robot_shortname, parent)
robot = model.create_user_robot(robot_shortname, parent)
log_action('create_robot', parent.username, {'robot': robot_shortname})
return robot_view(robot.username, password), 201
return robot.to_dict(), 201
@require_user_admin
@nickname('deleteUserRobot')
def delete(self, robot_shortname):
""" Delete an existing robot. """
parent = get_authenticated_user()
model.user.delete_robot(format_robot_username(parent.username, robot_shortname))
model.delete_robot(format_robot_username(parent.username, robot_shortname))
log_action('delete_robot', parent.username, {'robot': robot_shortname})
return '', 204
@ -123,6 +71,7 @@ class UserRobot(ApiResource):
@related_user_resource(UserRobotList)
class OrgRobotList(ApiResource):
""" Resource for listing an organization's robots. """
@require_scope(scopes.ORG_ADMIN)
@nickname('getOrgRobots')
@parse_args()
@ -145,15 +94,15 @@ class OrgRobotList(ApiResource):
@related_user_resource(UserRobot)
class OrgRobot(ApiResource):
""" Resource for managing an organization's robots. """
@require_scope(scopes.ORG_ADMIN)
@nickname('getOrgRobot')
def get(self, orgname, robot_shortname):
""" Returns the organization's robot with the specified name. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, password = model.user.get_robot(robot_shortname, parent)
return robot_view(robot.username, password)
robot = model.get_org_robot(robot_shortname, orgname)
return robot.to_dict()
raise Unauthorized()
@ -163,10 +112,9 @@ class OrgRobot(ApiResource):
""" Create a new robot in the organization. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, password = model.user.create_robot(robot_shortname, parent)
robot = model.create_org_robot(robot_shortname, orgname)
log_action('create_robot', orgname, {'robot': robot_shortname})
return robot_view(robot.username, password), 201
return robot.to_dict(), 201
raise Unauthorized()
@ -176,7 +124,7 @@ class OrgRobot(ApiResource):
""" Delete an existing organization robot. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
model.user.delete_robot(format_robot_username(orgname, robot_shortname))
model.delete_robot(format_robot_username(orgname, robot_shortname))
log_action('delete_robot', orgname, {'robot': robot_shortname})
return '', 204
@ -188,16 +136,17 @@ class OrgRobot(ApiResource):
'The short name for the robot, without any user or organization prefix')
class UserRobotPermissions(ApiResource):
""" Resource for listing the permissions a user's robot has in the system. """
@require_user_admin
@nickname('getUserRobotPermissions')
def get(self, robot_shortname):
""" Returns the list of repository permissions for the user's robot. """
parent = get_authenticated_user()
robot, _ = model.user.get_robot(robot_shortname, parent)
permissions = model.permission.list_robot_permissions(robot.username)
robot = model.get_user_robot(robot_shortname, parent)
permissions = model.list_robot_permissions(robot.username)
return {
'permissions': [permission_view(permission) for permission in permissions]
'permissions': [permission.to_dict() for permission in permissions]
}
@ -208,18 +157,18 @@ class UserRobotPermissions(ApiResource):
@related_user_resource(UserRobotPermissions)
class OrgRobotPermissions(ApiResource):
""" Resource for listing the permissions an org's robot has in the system. """
@require_user_admin
@nickname('getOrgRobotPermissions')
def get(self, orgname, robot_shortname):
""" Returns the list of repository permissions for the org's robot. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, _ = model.user.get_robot(robot_shortname, parent)
permissions = model.permission.list_robot_permissions(robot.username)
robot = model.get_org_robot(robot_shortname, orgname)
permissions = model.list_robot_permissions(robot.username)
return {
'permissions': [permission_view(permission) for permission in permissions]
'permissions': [permission.to_dict() for permission in permissions]
}
abort(403)
@ -230,14 +179,15 @@ class OrgRobotPermissions(ApiResource):
'The short name for the robot, without any user or organization prefix')
class RegenerateUserRobot(ApiResource):
""" Resource for regenerate an organization's robot's token. """
@require_user_admin
@nickname('regenerateUserRobotToken')
def post(self, robot_shortname):
""" Regenerates the token for a user's robot. """
parent = get_authenticated_user()
robot, password = model.user.regenerate_robot_token(robot_shortname, parent)
robot = model.regenerate_user_robot_token(robot_shortname, parent)
log_action('regenerate_robot_token', parent.username, {'robot': robot_shortname})
return robot_view(robot.username, password)
return robot.to_dict()
@resource('/v1/organization/<orgname>/robots/<robot_shortname>/regenerate')
@ -247,15 +197,15 @@ class RegenerateUserRobot(ApiResource):
@related_user_resource(RegenerateUserRobot)
class RegenerateOrgRobot(ApiResource):
""" Resource for regenerate an organization's robot's token. """
@require_scope(scopes.ORG_ADMIN)
@nickname('regenerateOrgRobotToken')
def post(self, orgname, robot_shortname):
""" Regenerates the token for an organization robot. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, password = model.user.regenerate_robot_token(robot_shortname, parent)
robot = model.regenerate_org_robot_token(robot_shortname, orgname)
log_action('regenerate_robot_token', orgname, {'robot': robot_shortname})
return robot_view(robot.username, password)
return robot.to_dict()
raise Unauthorized()

159
endpoints/api/robot_models_interface.py Normal file
View file

@ -0,0 +1,159 @@
from abc import ABCMeta, abstractmethod
from collections import namedtuple
from six import add_metaclass
class Permission(namedtuple('Permission', ['repository_name', 'repository_visibility_name', 'role_name'])):
"""
Permission the relationship between a robot and a repository and whether that robot can see the repo.
"""
def to_dict(self):
return {
'repository': {
'name': self.repository_name,
'is_public': self.repository_visibility_name == 'public'
},
'role': self.role_name
}
class Team(namedtuple('Team', ['name', 'avatar'])):
"""
Team represents a team entry for a robot list entry.
:type name: string
:type avatar: {string -> string}
"""
class RobotWithPermissions(
namedtuple('RobotWithPermissions', [
'name',
'password',
'teams',
'repository_names',
])):
"""
RobotWithPermissions is a list of robot entries.
:type name: string
:type password: string
:type teams: [Team]
:type repository_names: [string]
"""
def to_dict(self):
return {
'name': self.name,
'token': self.password,
'teams': [team.to_dict() for team in self.teams],
'repositories': self.repository_names
}
class Robot(
namedtuple('Robot', [
'name',
'password',
])):
"""
Robot represents a robot entity.
:type name: string
:type password: string
"""
def to_dict(self):
return {
'name': self.name,
'token': self.password
}
@add_metaclass(ABCMeta)
class RobotInterface(object):
"""
Interface that represents all data store interactions required by the Robot API
"""
@abstractmethod
def get_org_robot(self, robot_shortname, orgname):
"""
Returns:
Robot object
"""
@abstractmethod
def get_user_robot(self, robot_shortname, owning_user):
"""
Returns:
Robot object
"""
@abstractmethod
def create_user_robot(self, robot_shortname, owning_user):
"""
Returns:
Robot object
"""
@abstractmethod
def create_org_robot(self, robot_shortname, orgname):
"""
Returns:
Robot object
"""
@abstractmethod
def delete_robot(self, robot_username):
"""
Returns:
Robot object
"""
@abstractmethod
def regenerate_user_robot_token(self, robot_shortname, owning_user):
"""
Returns:
Robot object
"""
@abstractmethod
def regenerate_org_robot_token(self, robot_shortname, orgname):
"""
Returns:
Robot object
"""
@abstractmethod
def list_entity_robot_permission_teams(self, prefix, include_permissions=False):
"""
Returns:
list of RobotWithPermissions objects
"""
@abstractmethod
def list_robot_permissions(self, username):
"""
Returns:
list of Robot objects
"""

86
endpoints/api/robot_models_pre_oci.py Normal file
View file

@ -0,0 +1,86 @@
from app import avatar
from data import model
from data.database import User, FederatedLogin, Team, Repository
from endpoints.api.robot_models_interface import RobotInterface, Robot, RobotWithPermissions, Permission
class RobotPreOCIModel(RobotInterface):
def list_robot_permissions(self, username):
permissions = model.permission.list_robot_permissions(username)
return [Permission(permission.repository.name, permission.repository.visibility.name, permission.role.name) for
permission in permissions]
def list_entity_robot_permission_teams(self, prefix, include_permissions=False):
tuples = model.user.list_entity_robot_permission_teams(prefix,
include_permissions=include_permissions)
robots = {}
robot_teams = set()
for robot_tuple in tuples:
robot_name = robot_tuple.get(User.username)
if robot_name not in robots:
robot_dict = {
'name': robot_name,
'token': robot_tuple.get(FederatedLogin.service_ident),
}
if include_permissions:
robot_dict.update({
'teams': [],
'repositories': []
})
robots[robot_name] = RobotWithPermissions(robot_dict['name'], robot_dict['token'], robot_dict['teams'],
robot_dict['repositories'])
robots[robot_name] = Robot(robot_dict['name'], robot_dict['token'])
if include_permissions:
team_name = robot_tuple.get(Team.name)
repository_name = robot_tuple.get(Repository.name)
if team_name is not None:
check_key = robot_name + ':' + team_name
if check_key not in robot_teams:
robot_teams.add(check_key)
robot_dict['teams'].append(Team(
team_name,
avatar.get_data(team_name, team_name, 'team')
))
if repository_name is not None:
if repository_name not in robot_dict['repositories']:
robot_dict['repositories'].append(repository_name)
return robots.values()
def regenerate_user_robot_token(self, robot_shortname, owning_user):
robot, password = model.user.regenerate_robot_token(robot_shortname, owning_user)
return Robot(robot.username, password)
def regenerate_org_robot_token(self, robot_shortname, orgname):
parent = model.organization.get_organization(orgname)
robot, password = model.user.regenerate_robot_token(robot_shortname, parent)
return Robot(robot.username, password)
def delete_robot(self, robot_username):
model.user.delete_robot(robot_username)
def create_user_robot(self, robot_shortname, owning_user):
robot, password = model.user.create_robot(robot_shortname, owning_user)
return Robot(robot.username, password)
def create_org_robot(self, robot_shortname, orgname):
parent = model.organization.get_organization(orgname)
robot, password = model.user.create_robot(robot_shortname, parent)
return Robot(robot.username, password)
def get_org_robot(self, robot_shortname, orgname):
parent = model.organization.get_organization(orgname)
robot, password = model.user.get_robot(robot_shortname, parent)
return Robot(robot.username, password)
def get_user_robot(self, robot_shortname, owning_user):
robot, password = model.user.get_robot(robot_shortname, owning_user)
return Robot(robot.username, password)
pre_oci_model = RobotPreOCIModel()