13.0.2
This commit is contained in:
Vladislav Nesterov
@@ -1,4 +1,5 @@
|
||||
from copy import deepcopy
|
||||
from dataclasses import Field, MISSING, dataclass, fields, is_dataclass
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from functools import wraps
|
||||
@@ -6,19 +7,24 @@ from importlib.metadata import version as get_package_version
|
||||
import inspect
|
||||
import json
|
||||
import re
|
||||
from types import UnionType
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
ClassVar,
|
||||
Iterable,
|
||||
Literal,
|
||||
Optional,
|
||||
ParamSpec,
|
||||
TypeVar,
|
||||
cast,
|
||||
get_args,
|
||||
get_origin,
|
||||
)
|
||||
import time
|
||||
|
||||
from ansible.module_utils.basic import AnsibleModule, env_fallback
|
||||
import dacite
|
||||
from dynamix_sdk import BVSAuth, DECS3OAuth, Dynamix
|
||||
from dynamix_sdk import __name__ as SDK_PACKAGE_NAME
|
||||
from dynamix_sdk import exceptions as sdk_exceptions
|
||||
@@ -33,6 +39,152 @@ P = ParamSpec('P')
|
||||
R = TypeVar('R')
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AnsibleModuleParamSet:
|
||||
"""Dataclass that can generate AnsibleModule argument_spec."""
|
||||
|
||||
_ANSIBLE_TYPE_BY_PYTHON: ClassVar[dict[type, str]] = {
|
||||
bool: 'bool',
|
||||
int: 'int',
|
||||
str: 'str',
|
||||
}
|
||||
_ALLOWED_FIELD_METADATA_KEYS: ClassVar[frozenset[str]] = frozenset({'choices'}) # noqa: E501
|
||||
|
||||
@classmethod
|
||||
def get_arg_spec(cls) -> dict[str, Any]:
|
||||
return cls._dataclass_to_arg_spec(cls)
|
||||
|
||||
@classmethod
|
||||
def from_dict[T: AnsibleModuleParamSet](
|
||||
cls: type[T], data: dict[str, Any]
|
||||
) -> T:
|
||||
return dacite.from_dict(
|
||||
data_class=cls,
|
||||
data=data,
|
||||
config=dacite.Config(cast=[Enum]),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _dataclass_to_arg_spec(cls, model_cls: type) -> dict[str, Any]:
|
||||
spec: dict[str, Any] = {}
|
||||
for field in fields(model_cls):
|
||||
spec[field.name] = cls._field_to_arg_spec(
|
||||
field=field,
|
||||
model_cls=model_cls,
|
||||
)
|
||||
return spec
|
||||
|
||||
@classmethod
|
||||
def _field_to_arg_spec(
|
||||
cls,
|
||||
field: Field,
|
||||
model_cls: type,
|
||||
) -> dict[str, Any]:
|
||||
if field.default is None:
|
||||
raise ValueError(
|
||||
f'{model_cls.__name__}.{field.name}: '
|
||||
'default=None is not allowed'
|
||||
)
|
||||
|
||||
if get_origin(field.type) is UnionType:
|
||||
args = get_args(field.type)
|
||||
if len(args) != 2 or type(None) not in args:
|
||||
raise ValueError(
|
||||
f'{model_cls.__name__}.{field.name}: '
|
||||
f'unsupported annotation {field.type!r}: '
|
||||
'expected T | None'
|
||||
)
|
||||
param_type = args[0] if args[1] is type(None) else args[1]
|
||||
allows_none = True
|
||||
else:
|
||||
param_type = field.type
|
||||
allows_none = False
|
||||
has_default = (
|
||||
field.default is not MISSING
|
||||
or field.default_factory is not MISSING
|
||||
)
|
||||
|
||||
unknown_metadata_keys = (
|
||||
set(field.metadata) - cls._ALLOWED_FIELD_METADATA_KEYS
|
||||
)
|
||||
if unknown_metadata_keys:
|
||||
raise ValueError(
|
||||
f'{model_cls.__name__}.{field.name}: '
|
||||
f'unsupported metadata keys {sorted(unknown_metadata_keys)!r}'
|
||||
)
|
||||
|
||||
try:
|
||||
entry = cls._type_to_arg_spec(param_type)
|
||||
except ValueError as exc:
|
||||
raise ValueError(
|
||||
f'{model_cls.__name__}.{field.name}: {exc}'
|
||||
)
|
||||
|
||||
if entry['type'] == 'dict' and 'options' in entry:
|
||||
if not allows_none:
|
||||
entry['apply_defaults'] = True
|
||||
return entry
|
||||
|
||||
metadata_choices = field.metadata.get('choices')
|
||||
if metadata_choices is not None:
|
||||
entry['choices'] = list(metadata_choices)
|
||||
|
||||
if not allows_none and not has_default:
|
||||
entry['required'] = True
|
||||
elif has_default:
|
||||
entry['default'] = cls._resolve_default(field)
|
||||
|
||||
return entry
|
||||
|
||||
@classmethod
|
||||
def _type_to_arg_spec(cls, type_: Any) -> dict[str, Any]:
|
||||
if get_origin(type_) is list:
|
||||
list_args = get_args(type_)
|
||||
if len(list_args) != 1:
|
||||
raise ValueError(f'unsupported annotation {type_!r}')
|
||||
elem_spec = cls._type_to_arg_spec(list_args[0])
|
||||
entry: dict[str, Any] = {
|
||||
'type': 'list',
|
||||
'elements': elem_spec['type'],
|
||||
}
|
||||
if 'options' in elem_spec:
|
||||
entry['options'] = elem_spec['options']
|
||||
if 'choices' in elem_spec:
|
||||
entry['choices'] = elem_spec['choices']
|
||||
return entry
|
||||
|
||||
if isinstance(type_, type) and is_dataclass(type_):
|
||||
return {
|
||||
'type': 'dict',
|
||||
'options': cls._dataclass_to_arg_spec(type_),
|
||||
}
|
||||
if isinstance(type_, type) and issubclass(type_, Enum):
|
||||
return {
|
||||
'type': 'str',
|
||||
'choices': list(type_._member_names_),
|
||||
}
|
||||
if get_origin(type_) is Literal:
|
||||
return {
|
||||
'type': 'str',
|
||||
'choices': list(get_args(type_)),
|
||||
}
|
||||
if type_ in cls._ANSIBLE_TYPE_BY_PYTHON:
|
||||
return {'type': cls._ANSIBLE_TYPE_BY_PYTHON[type_]}
|
||||
raise ValueError(f'unsupported annotation {type_!r}')
|
||||
|
||||
@staticmethod
|
||||
def _resolve_default(field: Field) -> Any:
|
||||
if field.default is not MISSING:
|
||||
default = field.default
|
||||
elif field.default_factory is not MISSING:
|
||||
default = field.default_factory()
|
||||
else:
|
||||
raise ValueError(f'{field.name}: no default to resolve')
|
||||
if isinstance(default, Enum):
|
||||
return default.name
|
||||
return default
|
||||
|
||||
|
||||
class DecortController(object):
|
||||
"""DecortController is a utility class that holds target controller context and handles API requests formatting
|
||||
based on the requested authentication type.
|
||||
@@ -53,7 +205,7 @@ class DecortController(object):
|
||||
_api: sdk_types.API | None = None
|
||||
_usermanager_whoami_result: None | dict = None
|
||||
|
||||
ANSIBLE_MODULES_VERSION = '13.0.1'
|
||||
ANSIBLE_MODULES_VERSION = '13.0.2'
|
||||
COMPATIBLE_SDK_MINOR_VERSION = '2.0'
|
||||
|
||||
VM_RESIZE_NOT = 0
|
||||
|
||||
Reference in New Issue
Block a user