quay/endpoints/api/robot_models_pre_oci.py

import features

from app import avatar
from data import model
from active_migration import ActiveDataMigration, ERTMigrationFlags
from data.database import (User, FederatedLogin, RobotAccountToken, Team as TeamTable, Repository,
                           RobotAccountMetadata)
from endpoints.api.robot_models_interface import (RobotInterface, Robot, RobotWithPermissions, Team,
                                                  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_token=False,
                                         include_permissions=False, limit=None):
    tuples = model.user.list_entity_robot_permission_teams(prefix, limit=limit,
                                                           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:
        token = None
        if include_token:
          # TODO(remove-unenc): Remove branches once migrated.
          if robot_tuple.get(RobotAccountToken.token):
            token = robot_tuple.get(RobotAccountToken.token).decrypt()

          if token is None and ActiveDataMigration.has_flag(ERTMigrationFlags.READ_OLD_FIELDS):
            token = robot_tuple.get(FederatedLogin.service_ident)
            assert not token.startswith('robot:')

        robot_dict = {
          'name': robot_name,
          'token': token,
          'created': robot_tuple.get(User.creation_date),
          'last_accessed': (robot_tuple.get(User.last_accessed)
                            if features.USER_LAST_ACCESSED else None),
          'description': robot_tuple.get(RobotAccountMetadata.description),
          'unstructured_metadata': robot_tuple.get(RobotAccountMetadata.unstructured_json),
        }

        if include_permissions:
          robot_dict.update({
            'teams': [],
            'repositories': [],
          })

      robots[robot_name] = Robot(robot_dict['name'], robot_dict['token'], robot_dict['created'],
                                 robot_dict['last_accessed'], robot_dict['description'],
                                 robot_dict['unstructured_metadata'])
      if include_permissions:
        team_name = robot_tuple.get(TeamTable.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)
        robots[robot_name] = RobotWithPermissions(robot_dict['name'], robot_dict['token'],
                                                  robot_dict['created'],
                                                  (robot_dict['last_accessed'] 
                                                   if features.USER_LAST_ACCESSED else None),
                                                  robot_dict['teams'],
                                                  robot_dict['repositories'],
                                                  robot_dict['description'])

    return robots.values()

  def regenerate_user_robot_token(self, robot_shortname, owning_user):
    robot, password, metadata = model.user.regenerate_robot_token(robot_shortname, owning_user)
    return Robot(robot.username, password, robot.creation_date, robot.last_accessed,
                 metadata.description, metadata.unstructured_json)

  def regenerate_org_robot_token(self, robot_shortname, orgname):
    parent = model.organization.get_organization(orgname)
    robot, password, metadata = model.user.regenerate_robot_token(robot_shortname, parent)
    return Robot(robot.username, password, robot.creation_date, robot.last_accessed,
                 metadata.description, metadata.unstructured_json)

  def delete_robot(self, robot_username):
    model.user.delete_robot(robot_username)

  def create_user_robot(self, robot_shortname, owning_user, description, unstructured_metadata):
    robot, password = model.user.create_robot(robot_shortname, owning_user, description or '',
                                              unstructured_metadata)
    return Robot(robot.username, password, robot.creation_date, robot.last_accessed,
                 description or '', unstructured_metadata)

  def create_org_robot(self, robot_shortname, orgname, description, unstructured_metadata):
    parent = model.organization.get_organization(orgname)
    robot, password = model.user.create_robot(robot_shortname, parent, description or '',
                                              unstructured_metadata)
    return Robot(robot.username, password, robot.creation_date, robot.last_accessed,
                 description or '', unstructured_metadata)

  def get_org_robot(self, robot_shortname, orgname):
    parent = model.organization.get_organization(orgname)
    robot, password, metadata = model.user.get_robot_and_metadata(robot_shortname, parent)
    return Robot(robot.username, password, robot.creation_date, robot.last_accessed,
                 metadata.description, metadata.unstructured_json)

  def get_user_robot(self, robot_shortname, owning_user):
    robot, password, metadata = model.user.get_robot_and_metadata(robot_shortname, owning_user)
    return Robot(robot.username, password, robot.creation_date, robot.last_accessed,
                 metadata.description, metadata.unstructured_json)


pre_oci_model = RobotPreOCIModel()