quay/config_app/config_endpoints/api/superuser_models_interface.py

from abc import ABCMeta, abstractmethod
from collections import namedtuple
from six import add_metaclass

from config_app.config_endpoints.api import format_date


def user_view(user):
    return {
        'name': user.username,
        'kind': 'user',
        'is_robot': user.robot,
    }


class RepositoryBuild(namedtuple('RepositoryBuild',
                                 ['uuid', 'logs_archived', 'repository_namespace_user_username', 'repository_name',
                                  'can_write', 'can_read', 'pull_robot', 'resource_key', 'trigger', 'display_name',
                                  'started', 'job_config', 'phase', 'status', 'error', 'archive_url'])):
    """
    RepositoryBuild represents a build associated with a repostiory
    :type uuid: string
    :type logs_archived: boolean
    :type repository_namespace_user_username: string
    :type repository_name: string
    :type can_write: boolean
    :type can_write: boolean
    :type pull_robot: User
    :type resource_key: string
    :type trigger: Trigger
    :type display_name: string
    :type started: boolean
    :type job_config: {Any -> Any}
    :type phase: string
    :type status: string
    :type error: string
    :type archive_url: string
    """

    def to_dict(self):

        resp = {
            'id': self.uuid,
            'phase': self.phase,
            'started': format_date(self.started),
            'display_name': self.display_name,
            'status': self.status or {},
            'subdirectory': self.job_config.get('build_subdir', ''),
            'dockerfile_path': self.job_config.get('build_subdir', ''),
            'context': self.job_config.get('context', ''),
            'tags': self.job_config.get('docker_tags', []),
            'manual_user': self.job_config.get('manual_user', None),
            'is_writer': self.can_write,
            'trigger': self.trigger.to_dict(),
            'trigger_metadata': self.job_config.get('trigger_metadata', None) if self.can_read else None,
            'resource_key': self.resource_key,
            'pull_robot': user_view(self.pull_robot) if self.pull_robot else None,
            'repository': {
                'namespace': self.repository_namespace_user_username,
                'name': self.repository_name
            },
            'error': self.error,
        }

        if self.can_write:
            if self.resource_key is not None:
                resp['archive_url'] = self.archive_url
            elif self.job_config.get('archive_url', None):
                resp['archive_url'] = self.job_config['archive_url']

        return resp


class Approval(namedtuple('Approval', ['approver', 'approval_type', 'approved_date', 'notes'])):
    """
    Approval represents whether a key has been approved or not
    :type approver: User
    :type approval_type: string
    :type approved_date: Date
    :type notes: string
    """

    def to_dict(self):
        return {
            'approver': self.approver.to_dict() if self.approver else None,
            'approval_type': self.approval_type,
            'approved_date': self.approved_date,
            'notes': self.notes,
        }


class ServiceKey(namedtuple('ServiceKey', ['name', 'kid', 'service', 'jwk', 'metadata', 'created_date',
                                           'expiration_date', 'rotation_duration', 'approval'])):
    """
    ServiceKey is an apostille signing key
    :type name: string
    :type kid: int
    :type service: string
    :type jwk: string
    :type metadata: string
    :type created_date: Date
    :type expiration_date: Date
    :type rotation_duration: Date
    :type approval: Approval

    """

    def to_dict(self):
        return {
            'name': self.name,
            'kid': self.kid,
            'service': self.service,
            'jwk': self.jwk,
            'metadata': self.metadata,
            'created_date': self.created_date,
            'expiration_date': self.expiration_date,
            'rotation_duration': self.rotation_duration,
            'approval': self.approval.to_dict() if self.approval is not None else None,
        }


class User(namedtuple('User', ['username', 'email', 'verified', 'enabled', 'robot'])):
    """
    User represents a single user.
    :type username: string
    :type email: string
    :type verified: boolean
    :type enabled: boolean
    :type robot: User
    """

    def to_dict(self):
        user_data = {
            'kind': 'user',
            'name': self.username,
            'username': self.username,
            'email': self.email,
            'verified': self.verified,
            'enabled': self.enabled,
        }

        return user_data


class Organization(namedtuple('Organization', ['username', 'email'])):
    """
    Organization represents a single org.
    :type username: string
    :type email: string
    """

    def to_dict(self):
        return {
            'name': self.username,
            'email': self.email,
        }




@add_metaclass(ABCMeta)
class SuperuserDataInterface(object):
    """
    Interface that represents all data store interactions required by a superuser api.
    """
    @abstractmethod
    def list_all_service_keys(self):
        """
        Returns a list of service keys
        """