Merge pull request #2820 from charltonaustin/create_data_interface_for_subsystem_api/robot.py_630

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

endpoints/api/robot.py

@@ -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,
+  robots = model.list_entity_robot_permission_teams(prefix, include_permissions=include_permissions)
-                                                         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)
+    robot = model.get_user_robot(robot_shortname, parent)
-    return robot_view(robot.username, password)
+    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 = model.get_org_robot(robot_shortname, orgname)
-      robot, password = model.user.get_robot(robot_shortname, parent)
+      return robot.to_dict()
-      return robot_view(robot.username, password)
 
     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 = model.create_org_robot(robot_shortname, orgname)
-      robot, password = model.user.create_robot(robot_shortname, parent)
       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)
+    robot = model.get_user_robot(robot_shortname, parent)
-    permissions = model.permission.list_robot_permissions(robot.username)
+    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.get_org_robot(robot_shortname, orgname)
-      robot, _ = model.user.get_robot(robot_shortname, parent)
+      permissions = model.list_robot_permissions(robot.username)
-      permissions = model.permission.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 = model.regenerate_org_robot_token(robot_shortname, orgname)
-      robot, password = model.user.regenerate_robot_token(robot_shortname, parent)
       log_action('regenerate_robot_token', orgname, {'robot': robot_shortname})
-      return robot_view(robot.username, password)
+      return robot.to_dict()
 
     raise Unauthorized()

endpoints/api/robot_models_interface.py

@@ -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
+
+    """

endpoints/api/robot_models_pre_oci.py

@@ -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()