class DataSchema(Schema):
"""
implements data schema
[Schema](https://www.w3.org/TR/wot-thing-description11/#sec-data-schema-vocabulary-definition)
[Supported Fields](https://www.w3.org/TR/wot-thing-description11/#data-schema-fields)
"""
title: str = None
titles: Optional[dict[str, str]] = None
description: Optional[str] = None
descriptions: Optional[dict[str, str]] = None
const: Optional[bool] = None
default: Optional[Any] = None
readOnly: Optional[bool] = None
writeOnly: Optional[bool] = None # write only are to be considered actions with no return value
format: Optional[str] = None
unit: Optional[str] = None
type: Optional[str] = None
oneOf: Optional[list[JSON]] = None
model_config = ConfigDict(extra="allow")
_custom_schema_generators: ClassVar = dict()
def __init__(self):
super().__init__()
def ds_build_fields_from_property(self, property: Property) -> None:
"""populates schema information from descriptor object"""
assert isinstance(property, Property), f"only Property is a subclass of dataschema, given type: {type(property)}"
self.title = get_summary(property)
if property.constant:
self.const = property.constant
if property.readonly:
self.readOnly = property.readonly
if property.default is not None:
self.default = property.default
if property.doc:
self.description = Schema.format_doc(property.doc)
if self.title == self.description:
del self.title
if property.label is not None:
self.title = property.label
if property.metadata and property.metadata.get("unit", None) is not None:
self.unit = property.metadata["unit"]
if property.allow_None:
if not hasattr(self, 'oneOf') or self.oneOf is None:
self.oneOf = []
if hasattr(self, 'type') and self.type is not None:
self._move_own_type_to_oneOf()
if not any(types["type"] in [None, "null"] for types in self.oneOf):
self.oneOf.append(dict(type="null"))
# & _ds prefix is used to avoid name conflicts with PropertyAffordance class
# you dont know what you are building, whether the data schema or something else when viewed from property affordance
def ds_build_from_property(self, property: Property) -> None:
"""
generates the schema specific to the type,
calls `ds_build_fields_from_property()` after choosing the right type
"""
assert isinstance(property, Property)
if not isinstance(property, Property):
raise TypeError(f"Property affordance schema can only be generated for Property. "
f"Given type {type(property)}")
if isinstance(property, (String, Filename, Foldername, Path)):
data_schema = StringSchema()
elif isinstance(property, (Number, Integer)):
data_schema = NumberSchema()
elif isinstance(property, Boolean):
data_schema = BooleanSchema()
elif isinstance(property, (List, TypedList, Tuple, TupleSelector)):
data_schema = ArraySchema()
elif isinstance(property, Selector):
data_schema = EnumSchema()
elif isinstance(property, (TypedDict, TypedKeyMappingsDict)):
data_schema = ObjectSchema()
elif isinstance(property, ClassSelector):
data_schema = OneOfSchema()
elif self._custom_schema_generators.get(property, NotImplemented) is not NotImplemented:
data_schema = self._custom_schema_generators[property]()
elif isinstance(property, Property) and property.model is not None:
from .pydantic_extensions import GenerateJsonSchemaWithoutDefaultTitles, type_to_dataschema
base_data_schema = DataSchema()
base_data_schema._build_from_property(property=property)
if isinstance(property.model, dict):
given_data_schema = property.model
elif isinstance(property.model, (BaseModel, RootModel)):
given_data_schema = type_to_dataschema(property.model).model_dump(mode='json', exclude_none=True)
if base_data_schema.oneOf: # allow_None = True
base_data_schema.oneOf.append(given_data_schema)
else:
for key, value in given_data_schema.items():
setattr(base_data_schema, key, value)
data_schema = base_data_schema
else:
raise TypeError(f"WoT schema generator for this descriptor/property is not implemented. name {property.name} & type {type(property)}")
data_schema.ds_build_fields_from_property(property)
for field_name in data_schema.model_dump(exclude_unset=True).keys():
field_value = getattr(data_schema, field_name, NotImplemented)
if field_value is not NotImplemented:
setattr(self, field_name, field_value)
def _move_own_type_to_oneOf(self):
"""move type to oneOf"""
raise NotImplementedError("Implement this method in subclass for each data type")
def _model_to_dataschema():
def type_to_dataschema(t: Union[type, BaseModel], **kwargs) -> DataSchema:
"""Convert a Python type to a Thing Description DataSchema
This makes use of pydantic's `schema_of` function to create a
json schema, then applies some fixes to make a DataSchema
as per the Thing Description (because Thing Description is
almost but not quite compatible with JSONSchema).
Additional keyword arguments are added to the DataSchema,
and will override the fields generated from the type that
is passed in. Typically you'll want to use this for the
`title` field.
"""
if isinstance(t, BaseModel):
json_schema = t.model_json_schema()
else:
json_schema = TypeAdapter(t).json_schema()
schema_dict = jsonschema_to_dataschema(json_schema)
# Definitions of referenced ($ref) schemas are put in a
# key called "definitions" or "$defs" by pydantic. We should delete this.
# TODO: find a cleaner way to do this
# This shouldn't be a severe problem: we will fail with a
# validation error if other junk is left in the schema.
for k in ["definitions", "$defs"]:
if k in schema_dict:
del schema_dict[k]
schema_dict.update(kwargs)
try:
return DataSchema(**schema_dict)
except ValidationError as ve:
print(
"Error while constructing DataSchema from the "
"following dictionary:\n"
+ JSONSerializer().dumps(schema_dict, indent=2)
+ "Before conversion, the JSONSchema was:\n"
+ JSONSerializer().dumps(json_schema, indent=2)
)
raise ve