# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 OpenStack Foundation
# Copyright 2012 Grid Dynamics
# Copyright 2013 OpenStack Foundation
# All Rights Reserved.
#
# 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.
"""
Base utilities to build API operation managers and objects on top of.
"""
from __future__ import annotations
# E1102: %s is not callable
# pylint: disable=E1102
import abc
from collections.abc import Sequence
import copy
from http import client as http_client
from typing import Any
from typing import Callable
from typing import cast
from typing import ClassVar
from urllib import parse as urlparse
from oslo_utils import strutils
from ironicclient.common.apiclient import exceptions
from ironicclient.common.i18n import _
[docs]
def getid(obj: Any) -> Any:
"""Return id if argument is a Resource.
Abstracts the common pattern of allowing both an object or an object's ID
(UUID) as a parameter when dealing with relationships.
"""
try:
if obj.uuid:
return obj.uuid
except AttributeError:
pass
try:
return obj.id
except AttributeError:
return obj
# TODO(aababilov): call run_hooks() in HookableMixin's child classes
[docs]
class HookableMixin(object):
"""Mixin so classes can register and run hooks."""
_hooks_map: ClassVar[dict[str, list[Callable[..., Any]]]] = {}
[docs]
@classmethod
def add_hook(cls, hook_type: str, hook_func: Callable[..., Any]) -> None:
"""Add a new hook of specified type.
:param cls: class that registers hooks
:param hook_type: hook type, e.g., '__pre_parse_args__'
:param hook_func: hook function
"""
if hook_type not in cls._hooks_map:
cls._hooks_map[hook_type] = []
cls._hooks_map[hook_type].append(hook_func)
[docs]
@classmethod
def run_hooks(cls, hook_type: str, *args: Any, **kwargs: Any) -> None:
"""Run all hooks of specified type.
:param cls: class that registers hooks
:param hook_type: hook type, e.g., '__pre_parse_args__'
:param args: args to be passed to every hook function
:param kwargs: kwargs to be passed to every hook function
"""
hook_funcs = cls._hooks_map.get(hook_type) or []
for hook_func in hook_funcs:
hook_func(*args, **kwargs)
[docs]
class BaseManager(HookableMixin):
"""Basic manager type providing common operations.
Managers interact with a particular type of API (servers, flavors, images,
etc.) and provide CRUD operations for them.
"""
resource_class: type[Resource] | None = None
def __init__(self, client: Any) -> None:
"""Initializes BaseManager with `client`.
:param client: instance of BaseClient descendant for HTTP requests
"""
super(BaseManager, self).__init__()
self.client = client
def _list(
self,
url: str,
response_key: str | None = None,
obj_class: type[Resource] | None = None,
json: dict[str, Any] | None = None,
) -> list[Resource]:
"""List the collection.
:param url: a partial URL, e.g., '/servers'
:param response_key: the key to be looked up in response dictionary,
e.g., 'servers'. If response_key is None - all response body
will be used.
:param obj_class: class for constructing the returned objects
(self.resource_class will be used by default)
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
"""
if json:
body = self.client.post(url, json=json).json()
else:
body = self.client.get(url).json()
if obj_class is None:
obj_class = self.resource_class
if obj_class is None:
raise ValueError("resource_class must be set")
data = body[response_key] if response_key is not None else body
# NOTE(ja): keystone returns values as list as {'values': [ ... ]}
# unlike other services which just return the list...
try:
data = data["values"]
except (KeyError, TypeError):
pass
return [obj_class(self, res, loaded=True) for res in data if res]
def _get(self, url: str, response_key: str | None = None) -> Resource:
"""Get an object from collection.
:param url: a partial URL, e.g., '/servers'
:param response_key: the key to be looked up in response dictionary,
e.g., 'server'. If response_key is None - all response body
will be used.
"""
if self.resource_class is None:
raise ValueError("resource_class must be set")
body = self.client.get(url).json()
data = body[response_key] if response_key is not None else body
return self.resource_class(self, data, loaded=True)
def _head(self, url: str) -> bool:
"""Retrieve request headers for an object.
:param url: a partial URL, e.g., '/servers'
"""
resp = self.client.head(url)
return bool(resp.status_code == http_client.NO_CONTENT)
def _post(
self,
url: str,
json: dict[str, Any],
response_key: str | None = None,
return_raw: bool = False,
) -> dict[str, Any] | Resource:
"""Create an object.
:param url: a partial URL, e.g., '/servers'
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
:param response_key: the key to be looked up in response dictionary,
e.g., 'server'. If response_key is None - all response body
will be used.
:param return_raw: flag to force returning raw JSON instead of
Python object of self.resource_class
"""
if self.resource_class is None:
raise ValueError("resource_class must be set")
body = self.client.post(url, json=json).json()
data = body[response_key] if response_key is not None else body
if return_raw:
return cast(dict[str, Any], data)
return self.resource_class(self, data)
def _put(
self,
url: str,
json: dict[str, Any] | None = None,
response_key: str | None = None,
) -> Resource | None:
"""Update an object with PUT method.
:param url: a partial URL, e.g., '/servers'
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
:param response_key: the key to be looked up in response dictionary,
e.g., 'servers'. If response_key is None - all response body
will be used.
"""
if self.resource_class is None:
raise ValueError("resource_class must be set")
resp = self.client.put(url, json=json)
# PUT requests may not return a body
if resp.content:
body = resp.json()
if response_key is not None:
return self.resource_class(self, body[response_key])
else:
return self.resource_class(self, body)
return None
def _patch(
self,
url: str,
json: dict[str, Any] | None = None,
response_key: str | None = None,
) -> Resource:
"""Update an object with PATCH method.
:param url: a partial URL, e.g., '/servers'
:param json: data that will be encoded as JSON and passed in POST
request (GET will be sent by default)
:param response_key: the key to be looked up in response dictionary,
e.g., 'servers'. If response_key is None - all response body
will be used.
"""
if self.resource_class is None:
raise ValueError("resource_class must be set")
body = self.client.patch(url, json=json).json()
if response_key is not None:
return self.resource_class(self, body[response_key])
else:
return self.resource_class(self, body)
def _delete(self, url: str) -> Any:
"""Delete an object.
:param url: a partial URL, e.g., '/servers/my-server'
"""
return self.client.delete(url)
[docs]
class ManagerWithFind(BaseManager, metaclass=abc.ABCMeta):
"""Manager with additional `find()`/`findall()` methods."""
[docs]
@abc.abstractmethod
def list(self) -> list[Resource]:
pass
[docs]
def find(self, **kwargs: Any) -> Resource:
"""Find a single item with attributes matching ``**kwargs``.
This isn't very efficient: it loads the entire list then filters on
the Python side.
"""
if self.resource_class is None:
raise ValueError("resource_class must be set")
matches = self.findall(**kwargs)
num_matches = len(matches)
if num_matches == 0:
msg = _("No %(name)s matching %(args)s.") % {
"name": self.resource_class.__name__,
"args": kwargs,
}
raise exceptions.NotFound(msg)
elif num_matches > 1:
raise exceptions.NoUniqueMatch()
else:
return matches[0]
[docs]
def findall(self, **kwargs: Any) -> Sequence[Resource]:
"""Find all items with attributes matching ``**kwargs``.
This isn't very efficient: it loads the entire list then filters on
the Python side.
"""
found: list[Resource] = []
searches = kwargs.items()
items = self.list()
for obj in items:
try:
if all(
getattr(obj, attr) == value
for (attr, value) in searches
):
found.append(obj)
except AttributeError:
continue
return found
[docs]
class CrudManager(BaseManager):
"""Base manager class for manipulating entities.
Children of this class are expected to define a `collection_key` and `key`.
- `collection_key`: Usually a plural noun by convention (e.g. `entities`);
used to refer collections in both URL's (e.g. `/v3/entities`) and JSON
objects containing a list of member resources (e.g. `{'entities': [{},
{}, {}]}`).
- `key`: Usually a singular noun by convention (e.g. `entity`); used to
refer to an individual member of the collection.
"""
collection_key: str | None = None
key: str | None = None
[docs]
def build_url(self, base_url: str | None = None, **kwargs: Any) -> str:
"""Builds a resource URL for the given kwargs.
Given an example collection where `collection_key = 'entities'` and
`key = 'entity'`, the following URL's could be generated.
By default, the URL will represent a collection of entities, e.g.::
/entities
If kwargs contains an `entity_id`, then the URL will represent a
specific member, e.g.::
/entities/{entity_id}
:param base_url: if provided, the generated URL will be appended to it
"""
url = base_url if base_url is not None else ""
url += "/%s" % self.collection_key
# do we have a specific entity?
entity_id = kwargs.get("%s_id" % self.key)
if entity_id is not None:
url += "/%s" % entity_id
return url
def _filter_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]:
"""Drop null values and handle ids."""
for key, ref in kwargs.copy().items():
if ref is None:
kwargs.pop(key)
else:
if isinstance(ref, Resource):
kwargs.pop(key)
kwargs["%s_id" % key] = getid(ref)
return kwargs
[docs]
def create(self, **kwargs: Any) -> Resource:
if self.key is None:
raise ValueError("key must be set")
kwargs = self._filter_kwargs(kwargs)
result = self._post(
self.build_url(**kwargs),
{self.key: kwargs},
self.key,
)
if isinstance(result, dict):
raise ValueError("Expected Resource, got dict")
return result
[docs]
def get(self, **kwargs: Any) -> Resource:
kwargs = self._filter_kwargs(kwargs)
return self._get(self.build_url(**kwargs), self.key)
[docs]
def head(self, **kwargs: Any) -> bool:
kwargs = self._filter_kwargs(kwargs)
return self._head(self.build_url(**kwargs))
[docs]
def list(
self, base_url: str | None = None, **kwargs: Any
) -> list[Resource]:
"""List the collection.
:param base_url: if provided, the generated URL will be appended to it
"""
kwargs = self._filter_kwargs(kwargs)
return self._list(
"%(base_url)s%(query)s"
% {
"base_url": self.build_url(base_url=base_url, **kwargs),
"query": (
"?%s" % urlparse.urlencode(kwargs)
if kwargs else ""
),
},
self.collection_key,
)
[docs]
def put(
self, base_url: str | None = None, **kwargs: Any
) -> Resource | None:
"""Update an element.
:param base_url: if provided, the generated URL will be appended to it
"""
kwargs = self._filter_kwargs(kwargs)
return self._put(self.build_url(base_url=base_url, **kwargs))
[docs]
def update(self, **kwargs: Any) -> Resource:
kwargs = self._filter_kwargs(kwargs)
params = kwargs.copy()
if self.key is None:
raise ValueError("key must be set")
params.pop("%s_id" % self.key)
return self._patch(
self.build_url(**kwargs),
{self.key: params},
self.key,
)
[docs]
def delete(self, **kwargs: Any) -> Any:
kwargs = self._filter_kwargs(kwargs)
return self._delete(self.build_url(**kwargs))
[docs]
def find(self, base_url: str | None = None, **kwargs: Any) -> Resource:
"""Find a single item with attributes matching ``**kwargs``.
:param base_url: if provided, the generated URL will be appended to it
"""
if self.resource_class is None:
raise ValueError("resource_class must be set")
kwargs = self._filter_kwargs(kwargs)
rl = self._list(
"%(base_url)s%(query)s"
% {
"base_url": self.build_url(base_url=base_url, **kwargs),
"query": "?%s" % urlparse.urlencode(kwargs) if kwargs else "",
},
self.collection_key,
)
num = len(rl)
if num == 0:
msg = _("No %(name)s matching %(args)s.") % {
"name": self.resource_class.__name__,
"args": kwargs,
}
raise exceptions.NotFound(msg)
elif num > 1:
raise exceptions.NoUniqueMatch
else:
return rl[0]
[docs]
class Extension(HookableMixin):
"""Extension descriptor."""
SUPPORTED_HOOKS: tuple[str, ...] = (
"__pre_parse_args__", "__post_parse_args__"
)
manager_class: type[BaseManager] | None = None
def __init__(self, name: str, module: Any) -> None:
super(Extension, self).__init__()
self.name = name
self.module = module
self._parse_extension_module()
def _parse_extension_module(self) -> None:
self.manager_class = None
for attr_name, attr_value in self.module.__dict__.items():
if attr_name in self.SUPPORTED_HOOKS:
self.add_hook(attr_name, attr_value)
else:
try:
if issubclass(attr_value, BaseManager):
self.manager_class = attr_value
except TypeError:
pass
def __repr__(self) -> str:
return "<Extension '%s'>" % self.name
[docs]
class Resource(object):
"""Base class for OpenStack resources (tenant, user, etc.).
This is pretty much just a bag for attributes.
"""
HUMAN_ID: bool = False
NAME_ATTR: str = "name"
def __init__(
self, manager: BaseManager, info: dict[str, Any], loaded: bool = False
) -> None:
"""Populate and bind to a manager.
:param manager: BaseManager object
:param info: dictionary representing resource attributes
:param loaded: prevent lazy-loading if set to True
"""
self.manager = manager
self._info = info
self._add_details(info)
self._loaded = loaded
def __repr__(self) -> str:
reprkeys = sorted(
k for k in self.__dict__.keys() if k[0] != "_" and k != "manager"
)
info = ", ".join("%s=%s" % (k, getattr(self, k)) for k in reprkeys)
return "<%s %s>" % (self.__class__.__name__, info)
@property
def human_id(self) -> str | None:
"""Human-readable ID which can be used for bash completion."""
if self.HUMAN_ID:
name = getattr(self, self.NAME_ATTR, None)
if name is not None:
return str(strutils.to_slug(name))
return None
def _add_details(self, info: dict[str, Any]) -> None:
for k, v in info.items():
try:
setattr(self, k, v)
self._info[k] = v
except AttributeError:
# In this case we already defined the attribute on the class
pass
def __getattr__(self, k: str) -> Any:
if k not in self.__dict__:
# NOTE(bcwaldon): disallow lazy-loading if already loaded once
if not self.is_loaded():
self.get()
return self.__getattr__(k)
raise AttributeError(k)
else:
return self.__dict__[k]
[docs]
def get(self) -> None:
"""Support for lazy loading details.
Some clients, such as novaclient have the option to lazy load the
details, details which can be loaded with this function.
"""
# set_loaded() first ... so if we have to bail, we know we tried.
self.set_loaded(True)
if not hasattr(self.manager, "get"):
return
new = self.manager.get(self.id)
if new:
self._add_details(new._info)
self._add_details(
{"x_request_id": self.manager.client.last_request_id}
)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Resource):
return NotImplemented
# two resources of different types are not equal
if not isinstance(other, self.__class__):
return False
return self._info == other._info
[docs]
def is_loaded(self) -> bool:
return self._loaded
[docs]
def set_loaded(self, val: bool) -> None:
self._loaded = val
[docs]
def to_dict(self) -> dict[str, Any]:
return copy.deepcopy(self._info)
