in src/aaz_dev/swagger/model/schema/schema.py [0:0]
def to_cmd(self, builder, **kwargs):
if self.ref_instance is not None:
model = builder(self.ref_instance, support_cls_schema=True)
else:
model = builder.build_schema(self)
if isinstance(model, CMDArraySchemaBase):
if self.all_of is not None:
# inherit from allOf
if len(self.all_of) > 1:
raise exceptions.InvalidSwaggerValueError(
msg=f"Multiple allOf is not supported for `{model.type}` type schema",
key=self.traces, value=None
)
model = builder(self.all_of[0], support_cls_schema=True)
if self.items:
assert isinstance(self.items, (Schema, ReferenceSchema))
v = builder(self.items, in_base=True, support_cls_schema=True)
assert isinstance(v, CMDSchemaBase)
model.item = v
# freeze because array item is frozen
if not model.frozen and model.item.frozen:
model.frozen = True
if self.x_ms_identifiers:
model.identifiers = []
for identifier in self.x_ms_identifiers:
model.identifiers.append(identifier)
elif isinstance(model, CMDObjectSchemaBase):
# props
prop_dict = {}
if model.props is not None:
# inherit from $ref
for prop in model.props:
prop_dict[prop.name] = prop
if self.all_of:
# inherit from allOf
for item in self.all_of:
disc_parent = item.get_disc_parent()
if disc_parent is not None and builder.find_traces(item.ref_instance.traces):
# discriminator parent already in trace, break reference loop
continue
v = builder(item, in_base=True, support_cls_schema=False)
assert isinstance(v, CMDObjectSchemaBase)
if v.fmt:
model.fmt = v.fmt
if v.props:
for p in v.props:
prop_dict[p.name] = p
if disc_parent is not None and not builder.find_traces(disc_parent.traces):
# directly use child definition instead of polymorphism.
# So the value for discriminator property is const.
is_children = False
disc_prop = disc_parent.discriminator
for disc_value, disc_child in disc_parent.disc_children.items():
if disc_child == self:
prop_dict[disc_prop].const = True
prop_dict[disc_prop].default = CMDSchemaDefault()
prop_dict[disc_prop].default.value = disc_value
is_children = True
break
if not is_children and len(self.all_of) == 1 and \
model.props is None and model.additional_props is None:
# inherit from allOf as reference only
model.discriminators = v.discriminators
if v.additional_props:
model.additional_props = v.additional_props
if self.properties:
for name, p in self.properties.items():
assert isinstance(p, (Schema, ReferenceSchema))
v = builder(p, in_base=False, support_cls_schema=True)
if v is None:
# ignore by mutability
continue
assert isinstance(v, CMDSchema)
v.name = name
prop_dict[name] = v
if self.required:
for name in self.required:
if name in prop_dict:
# read_only property is not required
if prop_dict[name].read_only:
logger.info(f"Ignore requirement of the read only property: {name}")
continue
# because required property will not be included in a cls definition,
# so it's fine to update it in parent level when prop_dict[name] is a cls definition.
prop_dict[name].required = True
if MutabilityEnum.Create == builder.mutability:
# for create operation
# when a property is required, it's frozen status must be consisted with the defined schema.
# This can help to for empty object schema.
prop_dict[name].frozen = builder.frozen
# discriminators
if self.disc_children:
discriminators = []
assert self.discriminator is not None
disc_prop = self.discriminator
for disc_value, disc_child in self.disc_children.items():
if builder.find_traces(disc_child.traces):
# discriminator child already in trace, break reference loop
continue
disc = CMDObjectSchemaDiscriminator()
disc.property = disc_prop
disc.value = disc_value
if disc_prop not in prop_dict:
raise exceptions.InvalidSwaggerValueError(
msg="Discriminator Property don't exist",
key=self.traces,
value=[disc_prop, builder.mutability]
)
if not hasattr(prop_dict[disc_prop], "enum"):
raise exceptions.InvalidSwaggerValueError(
msg="Invalid Discriminator Property type",
key=self.traces,
value=[disc_prop, prop_dict[disc_prop].type]
)
# make sure discriminator value is an enum item of discriminator property
if prop_dict[disc_prop].enum is None:
prop_dict[disc_prop].enum = CMDSchemaEnum()
prop_dict[disc_prop].enum.items = []
exist_disc_value = False
for enum_item in prop_dict[disc_prop].enum.items:
if enum_item.value == disc_value:
exist_disc_value = True
if not exist_disc_value:
enum_item = CMDSchemaEnumItem()
enum_item.value = disc_value
prop_dict[disc_prop].enum.items.append(enum_item)
v = builder(disc_child, in_base=True, support_cls_schema=False)
assert isinstance(v, CMDObjectSchemaBase)
if v.frozen and prop_dict[disc_prop].frozen:
disc.frozen = True
if v.props:
disc.props = [prop for prop in v.props if prop.name not in prop_dict]
if v.discriminators:
disc.discriminators = v.discriminators
discriminators.append(disc)
if discriminators:
model.discriminators = discriminators
# convert special properties when self is an azure resource
if self.x_ms_azure_resource and prop_dict:
if 'id' in prop_dict and self.resource_id_templates and not prop_dict['id'].frozen:
id_prop = prop_dict['id']
if not isinstance(id_prop, CMDResourceIdSchema):
try:
assert isinstance(id_prop, CMDStringSchema)
except:
raise
raw_data = id_prop.to_native()
prop_dict['id'] = id_prop = CMDResourceIdSchema(raw_data=raw_data)
if len(self.resource_id_templates) == 1:
id_prop.fmt = CMDResourceIdFormat()
id_prop.fmt.template = [*self.resource_id_templates][0]
else:
err = exceptions.InvalidSwaggerValueError(
msg="Multi resource id templates error",
key=self.traces,
value=self.resource_id_templates
)
logger.warning(err)
if 'location' in prop_dict and not prop_dict['location'].frozen:
location_prop = prop_dict['location']
if not isinstance(location_prop, CMDResourceLocationSchema):
assert isinstance(location_prop, CMDStringSchema)
raw_data = location_prop.to_native()
prop_dict['location'] = CMDResourceLocationSchema(raw_data=raw_data)
if prop_dict:
if "userAssignedIdentities" in prop_dict and "type" in prop_dict:
if isinstance(model, CMDObjectSchema):
# Transfer to IdentityObjectSchema
model = CMDIdentityObjectSchema(model.to_native())
elif isinstance(model, CMDObjectSchemaBase):
# Transfer to IdentityObjectSchemaBase
model = CMDIdentityObjectSchemaBase(model.to_native())
else:
raise NotImplementedError()
model.props = []
for prop in prop_dict.values():
model.props.append(prop)
# additional properties
if self.additional_properties:
if isinstance(self.additional_properties, (Schema, ReferenceSchema)):
v = builder(self.additional_properties, in_base=True, support_cls_schema=True)
if v is not None:
assert isinstance(v, CMDSchemaBase)
model.additional_props = CMDObjectSchemaAdditionalProperties()
model.additional_props.item = v
# item is required, it's frozen status must be consisted with the defined schema.
# This can help to for empty object.
model.additional_props.item.frozen = builder.frozen
elif isinstance(self.additional_properties, bool) and self.additional_properties is True:
# Free-Form Objects
# additionalProperties: true
model.additional_props = CMDObjectSchemaAdditionalProperties()
model.additional_props.item = CMDAnyTypeSchemaBase()
if model.additional_props:
if builder.read_only:
model.additional_props.read_only = builder.read_only
if builder.frozen:
model.additional_props.frozen = builder.frozen
if self.x_ms_client_flatten and isinstance(model, CMDObjectSchema):
# client flatten can only be supported for CMDObjectSchema install of CMDObjectSchemaBase.
# Because CMDObjectSchemaBase will not link with argument
model.client_flatten = True
# when all additional_props and props and discriminators of model is frozen then this model is frozen
if not model.frozen and (model.additional_props or model.props or model.discriminators): # an empty object is not included
need_frozen = True
if model.additional_props:
if not model.additional_props.frozen:
need_frozen = False
if model.props:
for prop in model.props:
if not prop.frozen:
need_frozen = False
break
if model.discriminators:
for disc in model.discriminators:
if not disc.frozen:
need_frozen = False
break
# Note: model will always frozen when an unempty object without any props, additional_props or discriminators,
# If this property is required by parent schema, it will be updated in parent.
model.frozen = need_frozen
else:
if self.all_of is not None:
# inherit from allOf
if len(self.all_of) > 1:
raise exceptions.InvalidSwaggerValueError(
msg=f"Multiple allOf is not supported for `{model.type}` type schema",
key=self.traces, value=None
)
model = builder(self.all_of[0], support_cls_schema=True)
builder.setup_fmt(model, self)
builder.setup_enum(model, self)
builder.setup_default(model, self)
builder.setup_nullable(model, self)
if isinstance(model, CMDSchema):
builder.setup_description(model, self)
builder.setup_secret(model, self)
return model