# 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. import abc import typing as ty import warnings from requests import structures from openstack import format from openstack import warnings as os_warnings _SEEN_FORMAT = '{name}_seen' _T1 = ty.TypeVar('_T1') _T2 = ty.TypeVar('_T2') _T3 = ty.TypeVar('_T3', str, bool, int, float) # case 1: data_type is unset -> return value as-is @ty.overload def _convert_type( value: _T1, data_type: None, list_type: None = None, ) -> _T1: ... # case 2: data_type is primitive type -> return value as said primitive type @ty.overload def _convert_type( value: _T1, data_type: type[_T3], list_type: None = None, ) -> _T3: ... # case 3: data_type is list, no list_type -> return value as list of whatever # we got @ty.overload def _convert_type( value: _T1, data_type: type[list[ty.Any]], list_type: None = None, ) -> list[_T1]: ... # case 4: data_type is list, list_type is primitive type -> return value as # list of said primitive type @ty.overload def _convert_type( value: ty.Any, data_type: type[list[ty.Any]], list_type: type[_T3], ) -> list[_T3]: ... # case 5: data_type is dict or Resource -> return value as dict/Resource @ty.overload def _convert_type( value: ty.Any, data_type: type[dict[ty.Any, ty.Any]], list_type: None = None, ) -> dict[ty.Any, ty.Any]: ... # case 6: data_type is a Formatter -> return value after conversion @ty.overload def _convert_type( value: ty.Any, data_type: type[format.Formatter[type[_T2]]], list_type: None = None, ) -> _T2: ... def _convert_type( value: _T1, data_type: type[ _T3 | list[ty.Any] | dict[ty.Any, ty.Any] | format.Formatter[_T2], ] | None, list_type: type[_T3] | None = None, ) -> _T1 | _T3 | list[_T3] | list[_T1] | dict[ty.Any, ty.Any] | _T2: # This should allow handling list of dicts that have their own # Component type directly. See openstack/compute/v2/limits.py # and the RateLimit type for an example. if data_type is None: return value elif issubclass(data_type, list): if isinstance(value, list | set | tuple): if not list_type: return data_type(value) return [_convert_type(x, list_type) for x in value] elif list_type: return [_convert_type(value, list_type)] else: return [value] elif isinstance(value, data_type): return value elif issubclass(data_type, dict): if isinstance(value, dict): return data_type(**value) # TODO(stephenfin): This should be a warning/error return dict() elif issubclass(data_type, format.Formatter): return data_type.deserialize(value) elif issubclass(data_type, bool): return data_type(value) elif issubclass(data_type, int | float): if isinstance(value, int | float): return data_type(value) if isinstance(value, str): if issubclass(data_type, int) and value.isdigit(): return data_type(value) elif issubclass(data_type, float) and ( x.isdigit() for x in value.split() ): return data_type(value) return data_type() # at this point we expect to have a str and you can convert basically # anything to a string, but there could be untyped code out there passing # random monstrosities so we need the try-catch to be safe try: return data_type(value) except ValueError: return data_type() class _BaseComponent(abc.ABC): # The name this component is being tracked as in the Resource key: ty.ClassVar[str] # The class to be used for mappings _map_cls: ty.ClassVar[type[ty.MutableMapping[str, ty.Any]]] = dict name: str data_type: ty.Any | None default: ty.Any alias: str | None aka: str | None alternate_id: bool list_type: ty.Any | None coerce_to_default: bool deprecated: bool deprecation_reason: str | None def __init__( self, name: str, type: ty.Any | None = None, default: ty.Any = None, alias: str | None = None, aka: str | None = None, alternate_id: bool = False, list_type: ty.Any | None = None, coerce_to_default: bool = False, deprecated: bool = False, deprecation_reason: str | None = None, ): """A typed descriptor for a component that makes up a Resource :param name: The name this component exists as on the server :param type: The type this component is expected to be by the server. By default this is None, meaning any value you specify will work. If you specify type=dict and then set a component to a string, __set__ will fail, for example. :param default: Typically None, but any other default can be set. :param alias: If set, alternative attribute on object to return. :param aka: If set, additional name attribute would be available under. :param alternate_id: When `True`, this property is known internally as a value that can be sent with requests that require an ID but when `id` is not a name the Resource has. This is a relatively uncommon case, and this setting should only be used once per Resource. :param list_type: If type is `list`, list_type designates what the type of the elements of the list should be. :param coerce_to_default: If the Component is None or not present, force the given default to be used. If a default is not given but a type is given, construct an empty version of the type in question. :param deprecated: Indicates if the option is deprecated. If it is, we display a warning message to the user. :param deprecation_reason: Custom deprecation message. """ self.name = name self.data_type = type if type is not None and coerce_to_default and not default: self.default = type() else: self.default = default self.alias = alias self.aka = aka self.alternate_id = alternate_id self.list_type = list_type self.coerce_to_default = coerce_to_default self.deprecated = deprecated self.deprecation_reason = deprecation_reason def __get__( self, instance: object, owner: type[object] | None = None, ) -> ty.Any: if instance is None: return self attributes = getattr(instance, self.key) try: value = attributes[self.name] except KeyError: value = self.default if self.alias: # Resource attributes can be aliased to each other. If neither # of them exist, then simply doing a # getattr(instance, self.alias) here sends things into # infinite recursion (this _get method is what gets called # when getattr(instance) is called. # To combat that, we set a flag on the instance saying that # we have seen the current name, and we check before trying # to resolve the alias if there is already a flag set for that # alias name. We then remove the seen flag for ourselves after # we exit the alias getattr to clean up after ourselves for # the next time. alias_flag = _SEEN_FORMAT.format(name=self.alias) if not getattr(instance, alias_flag, False): seen_flag = _SEEN_FORMAT.format(name=self.name) # Prevent infinite recursion setattr(instance, seen_flag, True) value = getattr(instance, self.alias) delattr(instance, seen_flag) self.warn_if_deprecated_property(value) return value # self.data_type() should not be called on None objects. if value is None: return None # This warning are pretty intrusive. Every time attribute is accessed # a warning is being thrown. In neutron clients we have way too many # places that still refer to tenant_id even though they may also # properly support project_id. For now we silence tenant_id warnings. if self.name != "tenant_id": self.warn_if_deprecated_property(value) return _convert_type(value, self.data_type, self.list_type) @property def type(self) -> ty.Any | None: # deprecated alias proxy return self.data_type def warn_if_deprecated_property(self, value: ty.Any) -> None: deprecated = object.__getattribute__(self, 'deprecated') deprecation_reason = object.__getattribute__( self, 'deprecation_reason', ) if value and deprecated: warnings.warn( "The field {!r} has been deprecated. {}".format( self.name, deprecation_reason or "Avoid usage." ), os_warnings.RemovedFieldWarning, ) def __set__(self, instance: object, value: ty.Any) -> None: if self.coerce_to_default and value is None: value_ = self.default elif value != self.default: value_ = _convert_type(value, self.data_type, self.list_type) else: value_ = value attributes = getattr(instance, self.key) attributes[self.name] = value_ def __delete__(self, instance: object) -> None: try: attributes = getattr(instance, self.key) del attributes[self.name] except KeyError: pass class Body(_BaseComponent): """Body attributes""" key = "_body" class Header(_BaseComponent): """Header attributes""" key = "_header" _map_cls = structures.CaseInsensitiveDict class URI(_BaseComponent): """URI attributes""" key = "_uri" class Computed(_BaseComponent): """Computed attributes""" key = "_computed"