quay/endpoints/api/repository_models_interface.py

from abc import ABCMeta, abstractmethod
from collections import namedtuple, defaultdict

from datetime import datetime
from six import add_metaclass

import features
from endpoints.api import format_date


class RepositoryBaseElement(
    namedtuple('RepositoryBaseElement', [
      'namespace_name', 'repository_name', 'is_starred', 'is_public', 'kind_name', 'description',
      'namespace_user_organization', 'namespace_user_removed_tag_expiration_s', 'last_modified',
      'action_count', 'should_last_modified', 'should_popularity', 'should_is_starred'
    ])):
  """
  Repository a single quay repository
  :type namespace_name: string
  :type repository_name: string
  :type is_starred: boolean
  :type is_public: boolean
  :type kind_name: string
  :type description: string
  :type namespace_user_organization: boolean
  :type should_last_modified: boolean
  :type should_popularity: boolean
  :type should_is_starred: boolean
  """

  def to_dict(self):
    repo = {
      'namespace': self.namespace_name,
      'name': self.repository_name,
      'description': self.description,
      'is_public': self.is_public,
      'kind': self.kind_name,
    }

    if self.should_last_modified:
      repo['last_modified'] = self.last_modified

    if self.should_popularity:
      repo['popularity'] = float(self.action_count if self.action_count else 0)

    if self.should_is_starred:
      repo['is_starred'] = self.is_starred

    return repo


class ApplicationRepository(
    namedtuple('ApplicationRepository', ['repository_base_elements', 'channels', 'releases'])):
  """
  Repository a single quay repository
  :type repository_base_elements: RepositoryBaseElement
  :type channels: [Channel]
  :type releases: [Release]
  """

  def to_dict(self):
    repo_data = {
      'namespace': self.repository_base_elements.namespace_name,
      'name': self.repository_base_elements.repository_name,
      'kind': self.repository_base_elements.kind_name,
      'description': self.repository_base_elements.description,
      'is_public': self.repository_base_elements.is_public,
      'is_organization': self.repository_base_elements.namespace_user_organization,
      'is_starred': self.repository_base_elements.is_starred,
      'channels': [chan.to_dict() for chan in self.channels],
      'releases': [release.to_dict() for release in self.releases],
    }

    return repo_data


class ImageRepositoryRepository(
    namedtuple('NonApplicationRepository',
               ['repository_base_elements', 'tags', 'counts', 'badge_token', 'trust_enabled'])):
  """
  Repository a single quay repository
  :type repository_base_elements: RepositoryBaseElement
  :type tags: [Tag]
  :type counts: [count]
  :type badge_token: string
  :type trust_enabled: boolean
  """

  def to_dict(self):
    return {
      'namespace': self.repository_base_elements.namespace_name,
      'name': self.repository_base_elements.repository_name,
      'kind': self.repository_base_elements.kind_name,
      'description': self.repository_base_elements.description,
      'is_public': self.repository_base_elements.is_public,
      'is_organization': self.repository_base_elements.namespace_user_organization,
      'is_starred': self.repository_base_elements.is_starred,
      'tags': {tag.name: tag.to_dict()
               for tag in self.tags},
      'status_token': self.badge_token if not self.repository_base_elements.is_public else '',
      'trust_enabled': bool(features.SIGNING) and self.trust_enabled,
      'tag_expiration_s': self.repository_base_elements.namespace_user_removed_tag_expiration_s,
    }


class Repository(namedtuple('Repository', [
    'namespace_name',
    'repository_name',
])):
  """
  Repository a single quay repository
  :type namespace_name: string
  :type repository_name: string
  """


class Channel(namedtuple('Channel', ['name', 'linked_tag_name', 'linked_tag_lifetime_start'])):
  """
  Repository a single quay repository
  :type name: string
  :type linked_tag_name: string
  :type linked_tag_lifetime_start: string
  """

  def to_dict(self):
    return {
      'name': self.name,
      'release': self.linked_tag_name,
      'last_modified': format_date(datetime.fromtimestamp(self.linked_tag_lifetime_start / 1000)),
    }


class Release(
    namedtuple('Channel', ['name', 'lifetime_start', 'releases_channels_map'])):
  """
  Repository a single quay repository
  :type name: string
  :type last_modified: string
  :type releases_channels_map: {string -> string}
  """

  def to_dict(self):
    return {
      'name': self.name,
      'last_modified': format_date(datetime.fromtimestamp(self.lifetime_start / 1000)),
      'channels': self.releases_channels_map[self.name],
    }


class Tag(
    namedtuple('Tag', [
      'name', 'image_docker_image_id', 'image_aggregate_size', 'lifetime_start_ts',
      'tag_manifest_digest', 'lifetime_end_ts',
    ])):
  """
  :type name: string
  :type image_docker_image_id: string
  :type image_aggregate_size: int
  :type lifetime_start_ts: int
  :type lifetime_end_ts: int|None
  :type tag_manifest_digest: string

  """

  def to_dict(self):
    tag_info = {
      'name': self.name,
      'image_id': self.image_docker_image_id,
      'size': self.image_aggregate_size
    }

    if self.lifetime_start_ts > 0:
      last_modified = format_date(datetime.fromtimestamp(self.lifetime_start_ts))
      tag_info['last_modified'] = last_modified

    if self.lifetime_end_ts:
      expiration = format_date(datetime.fromtimestamp(self.lifetime_end_ts))
      tag_info['expiration'] = expiration

    if self.tag_manifest_digest is not None:
      tag_info['manifest_digest'] = self.tag_manifest_digest

    return tag_info


class Count(namedtuple('Count', ['date', 'count'])):
  """
    date: DateTime
    count: int
  """

  def to_dict(self):
    return {
      'date': self.date.isoformat(),
      'count': self.count,
    }


@add_metaclass(ABCMeta)
class RepositoryDataInterface(object):
  """
  Interface that represents all data store interactions required by a Repository.
  """

  @abstractmethod
  def get_repo(self, namespace_name, repository_name, user):
    """
    Returns a repository
    """

  @abstractmethod
  def repo_exists(self, namespace_name, repository_name):
    """
    Returns true if a repo exists and false if not
    """

  @abstractmethod
  def create_repo(self, namespace, name, creating_user, description, visibility='private',
                  repo_kind='image'):
    """
    Returns creates a new repo
    """

  @abstractmethod
  def get_repo_list(self, starred, user, repo_kind, namespace, username, public, page_token,
                    last_modified, popularity):
    """
    Returns a RepositoryBaseElement
    """

  @abstractmethod
  def set_repository_visibility(self, namespace_name, repository_name, visibility):
    """
    Sets a repository's visibility if it is found
    """

  @abstractmethod
  def set_trust(self, namespace_name, repository_name, trust):
    """
    Sets a repository's trust_enabled field if it is found
    """

  @abstractmethod
  def set_description(self, namespace_name, repository_name, description):
    """
    Sets a repository's description if it is found.
    """

  @abstractmethod
  def purge_repository(self, namespace_name, repository_name):
    """
    Removes a repository
    """

  @abstractmethod
  def check_repository_usage(self, user_name, plan_found):
    """
    Creates a notification for a user if they are over or under on their repository usage
    """