22
33import ipaddress
44from collections .abc import Callable
5- from typing import TYPE_CHECKING , Any , get_args
5+ from typing import TYPE_CHECKING , Any , NamedTuple , get_args
66
77from ..protocols_base import CoreNodeBase
88from ..uuidt import UUIDT
1313 from ..schema import AttributeSchemaAPI
1414
1515
16+ class _GraphQLPayloadAttribute (NamedTuple ):
17+ """Result of resolving an attribute value for a GraphQL mutation.
18+
19+ Attributes:
20+ payload: Key/value entries to include in the mutation payload
21+ (e.g. ``{"value": ...}`` or ``{"from_pool": ...}``).
22+ variables: GraphQL variable bindings for unsafe string values.
23+ needs_metadata: When ``True``, the payload needs to append property flags/objects
24+ """
25+
26+ payload : dict [str , Any ]
27+ variables : dict [str , Any ]
28+ needs_metadata : bool
29+
30+ def to_dict (self ) -> dict [str , Any ]:
31+ return {"data" : self .payload , "variables" : self .variables }
32+
33+ def add_properties (self , properties_flag : dict [str , Any ], properties_object : dict [str , str | None ]) -> None :
34+ if not self .needs_metadata :
35+ return
36+ for prop_name , prop in properties_flag .items ():
37+ self .payload [prop_name ] = prop
38+
39+ for prop_name , prop in properties_object .items ():
40+ self .payload [prop_name ] = prop
41+
42+
1643class Attribute :
1744 """Represents an attribute of a Node, including its schema, value, and properties."""
1845
@@ -25,8 +52,12 @@ def __init__(self, name: str, schema: AttributeSchemaAPI, data: Any | dict) -> N
2552 """
2653 self .name = name
2754 self ._schema = schema
55+ self ._from_pool : dict [str , Any ] | None = None
2856
29- if not isinstance (data , dict ) or "value" not in data :
57+ if isinstance (data , dict ) and "from_pool" in data :
58+ self ._from_pool = data .pop ("from_pool" )
59+ data .setdefault ("value" , None )
60+ elif not isinstance (data , dict ) or "value" not in data :
3061 data = {"value" : data }
3162
3263 self ._properties_flag = PROPERTIES_FLAG
@@ -76,38 +107,55 @@ def value(self, value: Any) -> None:
76107 self ._value = value
77108 self .value_has_been_mutated = True
78109
79- def _generate_input_data (self ) -> dict | None :
80- data : dict [str , Any ] = {}
81- variables : dict [str , Any ] = {}
82-
83- if self .value is None :
84- if self ._schema .optional and self .value_has_been_mutated :
85- data ["value" ] = None
86- return data
87-
88- if isinstance (self .value , str ):
89- if SAFE_VALUE .match (self .value ):
90- data ["value" ] = self .value
91- else :
92- var_name = f"value_{ UUIDT .new ().hex }
93- variables [var_name ] = self .value
94- data ["value" ] = f"${ var_name }
95- elif isinstance (self .value , get_args (IP_TYPES )):
96- data ["value" ] = self .value .with_prefixlen
97- elif isinstance (self .value , CoreNodeBase ) and self .value .is_resource_pool ():
98- data ["from_pool" ] = {"id" : self .value .id }
99- else :
100- data ["value" ] = self .value
101-
102- for prop_name in self ._properties_flag :
103- if getattr (self , prop_name ) is not None :
104- data [prop_name ] = getattr (self , prop_name )
110+ def _initialize_graphql_payload (self ) -> _GraphQLPayloadAttribute :
111+ """Resolve the attribute value into a GraphQL mutation payload object."""
105112
106- for prop_name in self ._properties_object :
107- if getattr (self , prop_name ) is not None :
108- data [prop_name ] = getattr (self , prop_name )._generate_input_data ()
113+ # Pool-based allocation (dict data or resource-pool node)
114+ if self ._from_pool is not None :
115+ return _GraphQLPayloadAttribute (payload = {"from_pool" : self ._from_pool }, variables = {}, needs_metadata = True )
116+ if isinstance (self .value , CoreNodeBase ) and self .value .is_resource_pool ():
117+ return _GraphQLPayloadAttribute (
118+ payload = {"from_pool" : {"id" : self .value .id }}, variables = {}, needs_metadata = True
119+ )
109120
110- return {"data" : data , "variables" : variables }
121+ # Null value
122+ if self .value is None :
123+ data = {"value" : None } if (self ._schema .optional and self .value_has_been_mutated ) else {}
124+ return _GraphQLPayloadAttribute (payload = data , variables = {}, needs_metadata = False )
125+
126+ # Unsafe strings need a variable binding to avoid injection
127+ if isinstance (self .value , str ) and not SAFE_VALUE .match (self .value ):
128+ var_name = f"value_{ UUIDT .new ().hex }
129+ return _GraphQLPayloadAttribute (
130+ payload = {"value" : f"${ var_name } },
131+ variables = {var_name : self .value },
132+ needs_metadata = True ,
133+ )
134+
135+ # Safe strings, IP types, and everything else
136+ value = self .value .with_prefixlen if isinstance (self .value , get_args (IP_TYPES )) else self .value
137+ return _GraphQLPayloadAttribute (payload = {"value" : value }, variables = {}, needs_metadata = True )
138+
139+ def _generate_input_data (self ) -> _GraphQLPayloadAttribute :
140+ """Build the input payload for a GraphQL mutation on this attribute.
141+
142+ Returns a ResolvedValue object, which contains all the data required.
143+ """
144+ graphql_payload = self ._initialize_graphql_payload ()
145+
146+ properties_flag : dict [str , Any ] = {
147+ property_name : getattr (self , property_name )
148+ for property_name in self ._properties_flag
149+ if getattr (self , property_name ) is not None
150+ }
151+ properties_object : dict [str , str | None ] = {
152+ property_name : getattr (self , property_name )._generate_input_data ()
153+ for property_name in self ._properties_object
154+ if getattr (self , property_name ) is not None
155+ }
156+ graphql_payload .add_properties (properties_flag , properties_object )
157+
158+ return graphql_payload
111159
112160 def _generate_query_data (self , property : bool = False , include_metadata : bool = False ) -> dict | None :
113161 data : dict [str , Any ] = {"value" : None }
@@ -128,7 +176,15 @@ def _generate_query_data(self, property: bool = False, include_metadata: bool =
128176 return data
129177
130178 def _generate_mutation_query (self ) -> dict [str , Any ]:
131- if isinstance ( self .value , CoreNodeBase ) and self . value . is_resource_pool ():
179+ if self .is_from_pool_attribute ():
132180 # If it points to a pool, ask for the value of the pool allocated resource
133181 return {self .name : {"value" : None }}
134182 return {}
183+
184+ def is_from_pool_attribute (self ) -> bool :
185+ """Check whether this attribute's value is sourced from a resource pool.
186+
187+ Returns:
188+ True if the attribute value is a resource pool node or was explicitly allocated from a pool.
189+ """
190+ return (isinstance (self .value , CoreNodeBase ) and self .value .is_resource_pool ()) or self ._from_pool is not None
0 commit comments