# coding: utf-8 """ Kubernetes No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator) The version of the OpenAPI document: release-1.37 Generated by OpenAPI Generator (https://openapi-generator.tech) Do not edit the class manually. """ # noqa: E501 from __future__ import annotations import pprint import re # noqa: F401 import json from collections.abc import Mapping as _Mapping from pydantic import AliasChoices, BaseModel, ConfigDict, Field, StrictBool, StrictStr from typing import Any, ClassVar, Dict, List, Optional, cast as _cast from kubernetes.client.models.v1_node_selector import V1NodeSelector from kubernetes.client.models.v1beta1_device_attribute import V1beta1DeviceAttribute from kubernetes.client.models.v1beta1_device_capacity import V1beta1DeviceCapacity from kubernetes.client.models.v1beta1_device_counter_consumption import V1beta1DeviceCounterConsumption from kubernetes.client.models.v1beta1_device_taint import V1beta1DeviceTaint from kubernetes.client.models.v1beta1_node_allocatable_resource import V1beta1NodeAllocatableResource from typing import Optional, Set from typing_extensions import Self from pydantic_core import to_jsonable_python _OPENAPI_GENERATOR_TO_DICT = "_openapi_generator_to_dict" def _get_openapi_to_dict(value: Any) -> Any: # Reciprocal function references preserve inherited generated methods # without reserving model member names. Checking both directions also # rejects overrides whose decorators copy function attributes. to_dict = getattr(value, "to_dict", None) type_to_dict = getattr(type(value), "to_dict", None) if ( not callable(to_dict) or getattr(to_dict, "__func__", None) is not type_to_dict ): return None openapi_to_dict = getattr( type_to_dict, _OPENAPI_GENERATOR_TO_DICT, None ) if ( not callable(openapi_to_dict) or getattr( openapi_to_dict, _OPENAPI_GENERATOR_TO_DICT, None, ) is not type_to_dict ): return None return openapi_to_dict def _to_legacy_item(value: Any, serialize: bool) -> Any: to_dict = getattr(value, "to_dict", None) if _get_openapi_to_dict(value) is not None and callable(to_dict): return to_dict(serialize=serialize) if callable(to_dict): return to_dict() return value def _to_legacy_value(value: Any, serialize: bool) -> Any: # python-legacy converted only immediate list elements or dictionary values: # https://github.com/OpenAPITools/openapi-generator/blob/c84b949df1a9ec04ba75989cb49b45c940c2f694/modules/openapi-generator/src/main/resources/python-legacy/model.mustache#L203-L231 if isinstance(value, list): return [_to_legacy_item(item, serialize) for item in value] if isinstance(value, dict): return { key: _to_legacy_item(item, serialize) for key, item in value.items() } return _to_legacy_item(value, serialize) def _to_openapi_value(value: Any) -> Any: if isinstance(value, list): return [_to_openapi_value(item) for item in value] if isinstance(value, dict): return {key: _to_openapi_value(item) for key, item in value.items()} to_openapi_dict = _get_openapi_to_dict(value) if to_openapi_dict is not None: return to_openapi_dict(value) to_dict = getattr(value, "to_dict", None) if callable(to_dict): return to_dict() return value class V1beta1BasicDevice(BaseModel): """ BasicDevice defines one device instance. Attributes: openapi_types (dict): The key is attribute name and the value is attribute type. attribute_map (dict): The key is attribute name and the value is json key in definition. """ # noqa: E501 all_nodes: Optional[StrictBool] = Field(default=None, validation_alias=AliasChoices("allNodes", "all_nodes"), serialization_alias="allNodes", description="AllNodes indicates that all nodes have access to the device. Must only be set if Spec.PerDeviceNodeSelection is set to true. At most one of NodeName, NodeSelector and AllNodes can be set.") allow_multiple_allocations: Optional[StrictBool] = Field(default=None, validation_alias=AliasChoices("allowMultipleAllocations", "allow_multiple_allocations"), serialization_alias="allowMultipleAllocations", description="AllowMultipleAllocations marks whether the device is allowed to be allocated to multiple DeviceRequests. If AllowMultipleAllocations is set to true, the device can be allocated more than once, and all of its capacity is consumable, regardless of whether the requestPolicy is defined or not.") attributes: Optional[Dict[str, V1beta1DeviceAttribute]] = Field(default=None, description="Attributes defines the set of attributes for this device. The name of each attribute must be unique in that set. The maximum number of attributes and capacities combined is 32.") binding_conditions: Optional[List[StrictStr]] = Field(default=None, validation_alias=AliasChoices("bindingConditions", "binding_conditions"), serialization_alias="bindingConditions", description="BindingConditions defines the conditions for proceeding with binding. All of these conditions must be set in the per-device status conditions with a value of True to proceed with binding the pod to the node while scheduling the pod. The maximum number of binding conditions is 4. The conditions must be a valid condition type string. This is a beta field and requires enabling the DRADeviceBindingConditions and DRAResourceClaimDeviceStatus feature gates.") binding_failure_conditions: Optional[List[StrictStr]] = Field(default=None, validation_alias=AliasChoices("bindingFailureConditions", "binding_failure_conditions"), serialization_alias="bindingFailureConditions", description="BindingFailureConditions defines the conditions for binding failure. They may be set in the per-device status conditions. If any is true, a binding failure occurred. The maximum number of binding failure conditions is 4. The conditions must be a valid condition type string. This is a beta field and requires enabling the DRADeviceBindingConditions and DRAResourceClaimDeviceStatus feature gates.") binds_to_node: Optional[StrictBool] = Field(default=None, validation_alias=AliasChoices("bindsToNode", "binds_to_node"), serialization_alias="bindsToNode", description="BindsToNode indicates if the usage of an allocation involving this device has to be limited to exactly the node that was chosen when allocating the claim. If set to true, the scheduler will set the ResourceClaim.Status.Allocation.NodeSelector to match the node where the allocation was made. This is a beta field and requires enabling the DRADeviceBindingConditions and DRAResourceClaimDeviceStatus feature gates.") capacity: Optional[Dict[str, V1beta1DeviceCapacity]] = Field(default=None, description="Capacity defines the set of capacities for this device. The name of each capacity must be unique in that set. The maximum number of attributes and capacities combined is 32.") consumes_counters: Optional[List[V1beta1DeviceCounterConsumption]] = Field(default=None, validation_alias=AliasChoices("consumesCounters", "consumes_counters"), serialization_alias="consumesCounters", description="ConsumesCounters defines a list of references to sharedCounters and the set of counters that the device will consume from those counter sets. There can only be a single entry per counterSet. The maximum number of device counter consumptions per device is 2.") node_allocatable_resources: Optional[Dict[str, V1beta1NodeAllocatableResource]] = Field(default=None, validation_alias=AliasChoices("nodeAllocatableResources", "node_allocatable_resources"), serialization_alias="nodeAllocatableResources", description="NodeAllocatableResources defines the mapping of node resources that are managed by the DRA driver exposing this device. This includes resources currently reported in v1.Node `status.allocatable` that are not extended resources (see https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#extended-resources). Examples include \"cpu\", \"memory\", \"ephemeral-storage\", and hugepages. In addition to standard requests made through the Pod `spec`, these resources can also be requested through claims and allocated by the DRA driver. For example, a CPU DRA driver might allocate exclusive CPUs or auxiliary node memory dependencies of an accelerator device. The keys of this map are the node-allocatable resource names (e.g., \"cpu\", \"memory\"). Extended resource names are not permitted as keys.") node_name: Optional[StrictStr] = Field(default=None, validation_alias=AliasChoices("nodeName", "node_name"), serialization_alias="nodeName", description="NodeName identifies the node where the device is available. Must only be set if Spec.PerDeviceNodeSelection is set to true. At most one of NodeName, NodeSelector and AllNodes can be set.") node_selector: Optional[V1NodeSelector] = Field(default=None, validation_alias=AliasChoices("nodeSelector", "node_selector"), serialization_alias="nodeSelector") taints: Optional[List[V1beta1DeviceTaint]] = Field(default=None, description="If specified, these are the driver-defined taints. The maximum number of taints is 16. If taints are set for any device in a ResourceSlice, then the maximum number of allowed devices per ResourceSlice is 64 instead of 128. This is a beta field and requires enabling the DRADeviceTaints feature gate.") openapi_types: ClassVar[Dict[str, str]] = { "all_nodes": "bool", "allow_multiple_allocations": "bool", "attributes": "Dict[str, V1beta1DeviceAttribute]", "binding_conditions": "List[str]", "binding_failure_conditions": "List[str]", "binds_to_node": "bool", "capacity": "Dict[str, V1beta1DeviceCapacity]", "consumes_counters": "List[V1beta1DeviceCounterConsumption]", "node_allocatable_resources": "Dict[str, V1beta1NodeAllocatableResource]", "node_name": "str", "node_selector": "V1NodeSelector", "taints": "List[V1beta1DeviceTaint]" } attribute_map: ClassVar[Dict[str, str]] = { "all_nodes": "allNodes", "allow_multiple_allocations": "allowMultipleAllocations", "attributes": "attributes", "binding_conditions": "bindingConditions", "binding_failure_conditions": "bindingFailureConditions", "binds_to_node": "bindsToNode", "capacity": "capacity", "consumes_counters": "consumesCounters", "node_allocatable_resources": "nodeAllocatableResources", "node_name": "nodeName", "node_selector": "nodeSelector", "taints": "taints" } __properties: ClassVar[List[str]] = ["allNodes", "allowMultipleAllocations", "attributes", "bindingConditions", "bindingFailureConditions", "bindsToNode", "capacity", "consumesCounters", "nodeAllocatableResources", "nodeName", "nodeSelector", "taints"] @classmethod def __preprocess_input_names( cls, obj: Any, remove_hidden_storage_names: bool = True, ) -> Any: if not isinstance(obj, _Mapping): return obj obj = dict(obj) if "allNodes" not in obj and "all_nodes" in obj: obj["allNodes"] = obj["all_nodes"] obj.pop("all_nodes", None) if "allowMultipleAllocations" not in obj and "allow_multiple_allocations" in obj: obj["allowMultipleAllocations"] = obj["allow_multiple_allocations"] obj.pop("allow_multiple_allocations", None) if "bindingConditions" not in obj and "binding_conditions" in obj: obj["bindingConditions"] = obj["binding_conditions"] obj.pop("binding_conditions", None) if "bindingFailureConditions" not in obj and "binding_failure_conditions" in obj: obj["bindingFailureConditions"] = obj["binding_failure_conditions"] obj.pop("binding_failure_conditions", None) if "bindsToNode" not in obj and "binds_to_node" in obj: obj["bindsToNode"] = obj["binds_to_node"] obj.pop("binds_to_node", None) if "consumesCounters" not in obj and "consumes_counters" in obj: obj["consumesCounters"] = obj["consumes_counters"] obj.pop("consumes_counters", None) if "nodeAllocatableResources" not in obj and "node_allocatable_resources" in obj: obj["nodeAllocatableResources"] = obj["node_allocatable_resources"] obj.pop("node_allocatable_resources", None) if "nodeName" not in obj and "node_name" in obj: obj["nodeName"] = obj["node_name"] obj.pop("node_name", None) if "nodeSelector" not in obj and "node_selector" in obj: obj["nodeSelector"] = obj["node_selector"] obj.pop("node_selector", None) return obj model_config = ConfigDict( validate_by_name=True, validate_by_alias=True, validate_assignment=True, extra="forbid", protected_namespaces=(), ) def to_str(self) -> str: """Returns the string representation of the model""" return pprint.pformat(self.to_dict()) def __repr__(self) -> str: """For print and pprint""" return self.to_str() def __eq__(self, other: object) -> bool: """Returns true if both objects are equal""" if not isinstance(other, V1beta1BasicDevice): return False return self.to_dict() == other.to_dict() def __ne__(self, other: object) -> bool: """Returns true if both objects are not equal""" if not isinstance(other, V1beta1BasicDevice): return True return not (self == other) def to_json(self) -> str: """Returns the JSON representation of the model using alias""" to_openapi_dict = _get_openapi_to_dict(self) if to_openapi_dict is not None: return json.dumps(to_jsonable_python(to_openapi_dict(self))) return json.dumps(to_jsonable_python(self.to_dict())) @classmethod def from_json(cls, json_str: str) -> Optional[Self]: """Create an instance of V1beta1BasicDevice from a JSON string""" return cls.from_dict(json.loads(json_str)) def to_dict(self, serialize: bool = False) -> Dict[str, Any]: """Return all declared model fields using public or wire names.""" return { ("allNodes" if serialize else "all_nodes"): _to_legacy_value(getattr(self, "all_nodes", None), serialize), ("allowMultipleAllocations" if serialize else "allow_multiple_allocations"): _to_legacy_value(getattr(self, "allow_multiple_allocations", None), serialize), ("attributes" if serialize else "attributes"): _to_legacy_value(getattr(self, "attributes", None), serialize), ("bindingConditions" if serialize else "binding_conditions"): _to_legacy_value(getattr(self, "binding_conditions", None), serialize), ("bindingFailureConditions" if serialize else "binding_failure_conditions"): _to_legacy_value(getattr(self, "binding_failure_conditions", None), serialize), ("bindsToNode" if serialize else "binds_to_node"): _to_legacy_value(getattr(self, "binds_to_node", None), serialize), ("capacity" if serialize else "capacity"): _to_legacy_value(getattr(self, "capacity", None), serialize), ("consumesCounters" if serialize else "consumes_counters"): _to_legacy_value(getattr(self, "consumes_counters", None), serialize), ("nodeAllocatableResources" if serialize else "node_allocatable_resources"): _to_legacy_value(getattr(self, "node_allocatable_resources", None), serialize), ("nodeName" if serialize else "node_name"): _to_legacy_value(getattr(self, "node_name", None), serialize), ("nodeSelector" if serialize else "node_selector"): _to_legacy_value(getattr(self, "node_selector", None), serialize), ("taints" if serialize else "taints"): _to_legacy_value(getattr(self, "taints", None), serialize), } def __openapi_generator_modern_projection(self) -> Dict[str, Any]: """Return the dictionary representation of the model using alias. This has the following differences from calling pydantic's `self.model_dump(by_alias=True)`: * `None` is only added to the output dict for nullable fields that were set at model initialization. Other fields with value `None` are ignored. """ excluded_fields: Set[str] = set([ ]) _dict = self.model_dump( by_alias=True, exclude=excluded_fields, exclude_none=True, ) # override the default output from pydantic by calling `to_dict()` of each value in attributes (dict) _field_dict = {} if self.attributes: for _key_attributes in self.attributes: _field_dict[_key_attributes] = _to_openapi_value(self.attributes[_key_attributes]) if self.attributes[_key_attributes] is not None else None _dict['attributes'] = _field_dict # override the default output from pydantic by calling `to_dict()` of each value in capacity (dict) _field_dict = {} if self.capacity: for _key_capacity in self.capacity: _field_dict[_key_capacity] = _to_openapi_value(self.capacity[_key_capacity]) if self.capacity[_key_capacity] is not None else None _dict['capacity'] = _field_dict # override the default output from pydantic by calling `to_dict()` of each item in consumes_counters (list) _items = [] if self.consumes_counters: for _item_consumes_counters in self.consumes_counters: _items.append(_to_openapi_value(_item_consumes_counters) if _item_consumes_counters is not None else None) _dict['consumesCounters'] = _items # override the default output from pydantic by calling `to_dict()` of each value in node_allocatable_resources (dict) _field_dict = {} if self.node_allocatable_resources: for _key_node_allocatable_resources in self.node_allocatable_resources: _field_dict[_key_node_allocatable_resources] = _to_openapi_value(self.node_allocatable_resources[_key_node_allocatable_resources]) if self.node_allocatable_resources[_key_node_allocatable_resources] is not None else None _dict['nodeAllocatableResources'] = _field_dict # override the default output from pydantic by calling `to_dict()` of node_selector if self.node_selector: _dict['nodeSelector'] = _to_openapi_value(self.node_selector) # override the default output from pydantic by calling `to_dict()` of each item in taints (list) _items = [] if self.taints: for _item_taints in self.taints: _items.append(_to_openapi_value(_item_taints) if _item_taints is not None else None) _dict['taints'] = _items return _dict setattr( to_dict, _OPENAPI_GENERATOR_TO_DICT, __openapi_generator_modern_projection, ) setattr( __openapi_generator_modern_projection, _OPENAPI_GENERATOR_TO_DICT, to_dict, ) del __openapi_generator_modern_projection @classmethod def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: """Create an instance of V1beta1BasicDevice from a dict""" if obj is None: return None if not isinstance(obj, dict): return cls.model_validate(obj) obj = _cast( Dict[str, Any], cls.__preprocess_input_names( obj, remove_hidden_storage_names=True, ), ) _obj = cls.model_validate({ "allNodes": obj.get("allNodes"), "allowMultipleAllocations": obj.get("allowMultipleAllocations"), "attributes": dict( (_k, V1beta1DeviceAttribute.from_dict(_v)) for _k, _v in obj["attributes"].items() ) if obj.get("attributes") is not None else None, "bindingConditions": obj.get("bindingConditions"), "bindingFailureConditions": obj.get("bindingFailureConditions"), "bindsToNode": obj.get("bindsToNode"), "capacity": dict( (_k, V1beta1DeviceCapacity.from_dict(_v)) for _k, _v in obj["capacity"].items() ) if obj.get("capacity") is not None else None, "consumesCounters": [V1beta1DeviceCounterConsumption.from_dict(_item) for _item in obj["consumesCounters"]] if obj.get("consumesCounters") is not None else None, "nodeAllocatableResources": dict( (_k, V1beta1NodeAllocatableResource.from_dict(_v)) for _k, _v in obj["nodeAllocatableResources"].items() ) if obj.get("nodeAllocatableResources") is not None else None, "nodeName": obj.get("nodeName"), "nodeSelector": V1NodeSelector.from_dict(obj["nodeSelector"]) if obj.get("nodeSelector") is not None else None, "taints": [V1beta1DeviceTaint.from_dict(_item) for _item in obj["taints"]] if obj.get("taints") is not None else None }) return _obj