284 lines
11 KiB
Python
284 lines
11 KiB
Python
# Copyright 2017 Red Hat, Inc
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
__all__ = ['ServiceTypes']
|
|
|
|
import copy
|
|
|
|
import os_service_types.data
|
|
from os_service_types import exc
|
|
|
|
BUILTIN_DATA = os_service_types.data.read_data('service-types.json')
|
|
SERVICE_TYPES_URL = "https://service-types.openstack.org/service-types.json"
|
|
|
|
|
|
def _normalize_type(service_type):
|
|
return service_type.replace('_', '-')
|
|
|
|
|
|
class ServiceTypes(object):
|
|
"""Encapsulation of the OpenStack Service Types Authority data.
|
|
|
|
The Service Types Authority data will be either pulled from its remote
|
|
location or from local files as is appropriate.
|
|
|
|
If the user passes a Session, remote data will be fetched. If the user
|
|
does not do that, local builtin data will be used.
|
|
|
|
:param session: An object that behaves like a `requests.sessions.Session`
|
|
or a `keystoneauth1.session.Session` that provides a get method
|
|
and returns an object that behaves like a `requests.models.Response`.
|
|
Optional. If session is omitted, no remote actions will be performed.
|
|
:param bool only_remote: By default if there is a problem fetching data
|
|
remotely the builtin data will be returned as a fallback. only_remote
|
|
will cause remote failures to raise an error instead of falling back.
|
|
Optional, defaults to False.
|
|
:param bool warn: Emit warnings when a non-official service_type is
|
|
provided. This provides an easy way for consuming applications to
|
|
warn users when they are using old types.
|
|
:raises ValueError: If session is None and only_remote is True
|
|
:raises IOError: If session is given and only_remote is True and there is
|
|
an error fetching remote data.
|
|
"""
|
|
|
|
def __init__(self, session=None, only_remote=False, warn=False):
|
|
if not session and only_remote:
|
|
raise ValueError(
|
|
"only_remote was requested but no Session was provided.")
|
|
self._service_types_data = BUILTIN_DATA
|
|
self._warn = warn
|
|
if session:
|
|
try:
|
|
response = session.get(SERVICE_TYPES_URL)
|
|
response.raise_for_status()
|
|
self._service_types_data = response.json()
|
|
except IOError:
|
|
# If we can't fetch, fall backto BUILTIN
|
|
if only_remote:
|
|
raise
|
|
|
|
def _canonical_project_name(self, name):
|
|
"Convert repo name to project name."
|
|
if name is None:
|
|
raise ValueError("Empty project name is not allowed")
|
|
# Handle openstack/ prefix going away from STA data
|
|
return name.rpartition('/')[-1]
|
|
|
|
@property
|
|
def url(self):
|
|
"The URL from which the data was retrieved."
|
|
return SERVICE_TYPES_URL
|
|
|
|
@property
|
|
def version(self):
|
|
"The version of the data."
|
|
return self._service_types_data['version']
|
|
|
|
@property
|
|
def forward(self):
|
|
"Mapping service-type names to their aliases."
|
|
return copy.deepcopy(self._service_types_data['forward'])
|
|
|
|
@property
|
|
def reverse(self):
|
|
"Mapping aliases to their service-type names."
|
|
return copy.deepcopy(self._service_types_data['reverse'])
|
|
|
|
@property
|
|
def services(self):
|
|
"Full service-type data listing."
|
|
return copy.deepcopy(self._service_types_data['services'])
|
|
|
|
@property
|
|
def all_types_by_service_type(self):
|
|
"Mapping of official service type to official type and aliases."
|
|
return copy.deepcopy(
|
|
self._service_types_data['all_types_by_service_type'])
|
|
|
|
@property
|
|
def primary_service_by_project(self):
|
|
"Mapping of project name to the primary associated service."
|
|
return copy.deepcopy(
|
|
self._service_types_data['primary_service_by_project'])
|
|
|
|
@property
|
|
def service_types_by_project(self):
|
|
"Mapping of project name to a list of all associated service-types."
|
|
return copy.deepcopy(
|
|
self._service_types_data['service_types_by_project'])
|
|
|
|
def get_official_service_data(self, service_type):
|
|
"""Get the service data for an official service_type.
|
|
|
|
:param str service_type: The official service-type to get data for.
|
|
:returns dict: Service data for the service or None if not found.
|
|
"""
|
|
service_type = _normalize_type(service_type)
|
|
for service in self._service_types_data['services']:
|
|
if service_type == service['service_type']:
|
|
return service
|
|
return None
|
|
|
|
def get_service_data(self, service_type):
|
|
"""Get the service data for a given service_type.
|
|
|
|
:param str service_type: The service-type or alias to get data for.
|
|
:returns dict: Service data for the service or None if not found.
|
|
"""
|
|
service_type = self.get_service_type(service_type)
|
|
if not service_type:
|
|
return None
|
|
return self.get_official_service_data(service_type)
|
|
|
|
def is_official(self, service_type):
|
|
"""Is the given service-type an official service-type?
|
|
|
|
:param str service_type: The service-type to test.
|
|
:returns bool: True if it's an official type, False otherwise.
|
|
"""
|
|
return self.get_official_service_data(service_type) is not None
|
|
|
|
def is_alias(self, service_type):
|
|
"""Is the given service-type an alias?
|
|
|
|
:param str service_type: The service-type to test.
|
|
:returns bool: True if it's an alias type, False otherwise.
|
|
"""
|
|
service_type = _normalize_type(service_type)
|
|
return service_type in self._service_types_data['reverse']
|
|
|
|
def is_known(self, service_type):
|
|
"""Is the given service-type an official type or an alias?
|
|
|
|
:param str service_type: The service-type to test.
|
|
:returns bool: True if it's a known type, False otherwise.
|
|
"""
|
|
return self.is_official(service_type) or self.is_alias(service_type)
|
|
|
|
def is_match(self, requested, found):
|
|
"""Does a requested service-type match one found in the catalog?
|
|
|
|
A requested service-type matches a service-type in the catalog if
|
|
it is either a direct match, if the service-type in the catalog is
|
|
an official type and the requested type is one of its aliases, or
|
|
if the requested type is an official type and the type in the catalog
|
|
is one of its aliases.
|
|
|
|
A requested alias cannot match a different alias because there are
|
|
historical implications related to versioning to some historical
|
|
aliases that cannot be safely reasoned about in an automatic fashion.
|
|
|
|
:param str requested: A service-type that someone is looking for.
|
|
:param str found: A service-type found in a catalog
|
|
|
|
:returns bool: True if the service-type being requested matches the
|
|
entry in the catalog. False if it does not.
|
|
"""
|
|
# Exact match
|
|
if requested == found:
|
|
return True
|
|
|
|
# Found is official type, requested is one of its aliases
|
|
if requested in self.get_aliases(found):
|
|
return True
|
|
|
|
# Found is an alias, requested is an official type
|
|
if requested == self.get_service_type(found):
|
|
return True
|
|
|
|
return False
|
|
|
|
def get_aliases(self, service_type):
|
|
"""Returns the list of aliases for a given official service-type.
|
|
|
|
:param str service_type: An official service-type.
|
|
:returns list: List of aliases, or empty list if there are none.
|
|
"""
|
|
service_type = _normalize_type(service_type)
|
|
return self._service_types_data['forward'].get(service_type, [])
|
|
|
|
def get_service_type(self, service_type, permissive=False):
|
|
"""Given a possible service_type, return the official type.
|
|
|
|
:param str service_type: A potential service-type.
|
|
:param bool permissive:
|
|
Return the original type if the given service_type is not found.
|
|
:returns str: The official service-type, or None if there is no match.
|
|
"""
|
|
service_type = _normalize_type(service_type)
|
|
if self.is_official(service_type):
|
|
return service_type
|
|
official = self._service_types_data['reverse'].get(service_type)
|
|
if permissive and official is None:
|
|
if self._warn:
|
|
exc.warn(
|
|
exc.UnofficialUsageWarning,
|
|
given=service_type)
|
|
return service_type
|
|
if self._warn:
|
|
exc.warn(
|
|
exc.AliasUsageWarning, given=service_type, official=official)
|
|
return official
|
|
|
|
def get_all_types(self, service_type):
|
|
"""Get a list of official types and all known aliases.
|
|
|
|
:param str service_type: The service-type or alias to get data for.
|
|
:returns dict: Service data for the service or None if not found.
|
|
"""
|
|
service_type = _normalize_type(service_type)
|
|
if not self.is_known(service_type):
|
|
return [service_type]
|
|
return self.all_types_by_service_type[
|
|
self.get_service_type(service_type)]
|
|
|
|
def get_project_name(self, service_type):
|
|
"""Return the OpenStack project name for a given service_type.
|
|
|
|
:param str service_type: An official service-type or alias.
|
|
:returns str: The OpenStack project name or None if there is no match.
|
|
"""
|
|
service_type = _normalize_type(service_type)
|
|
service = self.get_service_data(service_type)
|
|
if service:
|
|
return service['project']
|
|
return None
|
|
|
|
def get_service_data_for_project(self, project_name):
|
|
"""Return the service information associated with a project.
|
|
|
|
:param name: A repository or project name in the form
|
|
``'openstack/{project}'`` or just ``'{project}'``.
|
|
:type name: str
|
|
:raises ValueError: If project_name is None
|
|
:returns: dict or None if not found
|
|
"""
|
|
project_name = self._canonical_project_name(project_name)
|
|
return self.primary_service_by_project.get(project_name)
|
|
|
|
def get_all_service_data_for_project(self, project_name):
|
|
"""Return the information for every service associated with a project.
|
|
|
|
:param name: A repository or project name in the form
|
|
``'openstack/{project}'`` or just ``'{project}'``.
|
|
:type name: str
|
|
:raises ValueError: If project_name is None
|
|
:returns: list of dicts
|
|
"""
|
|
data = []
|
|
for service_type in self.service_types_by_project.get(
|
|
project_name, []):
|
|
data.append(self.get_service_data(service_type))
|
|
return data
|