Coverage for filip/utils/model_generation.py: 38%

1""" 

2Code generator for data models from schema.json descriptions 

3""" 

4 

5import json 

6import shutil 

7from pathlib import Path 

8from tempfile import TemporaryDirectory 

9from typing import Union, Dict, Any, Type 

10from urllib import parse 

11from uuid import uuid4 

12from datamodel_code_generator import InputFileType, generate, ParseResult 

13from pydantic import create_model 

14 

15from filip.models.ngsi_v2.context import ContextAttribute, ContextEntity 

16 

17 

18def create_data_model_file( 

19 *, 

20 path: Union[Path, str], 

21 url: str = None, 

22 schema: Union[Path, str, ParseResult] = None, 

23 schema_type: Union[str, InputFileType] = InputFileType.JsonSchema, 

24 class_name: str = None, 

25) -> None: 

26 """ 

27 This will create a data model from data model definitions. The schemas 

28 can either downloaded from a url or passed as str or dict. Allowed input 

29 types are defined but the underlying toolbox. 

30 

31 Many data models suited for FIWARE are located here: 

32 https://github.com/smart-data-models/data-models 

33 

34 Args: 

35 path: 

36 path where the generated code should saved 

37 url: 

38 url to download the definition from 

39 schema_type (str): 

40 `auto`, `openapi`, `jsonschema`, `json`, `yaml`, `dict`, `csv` 

41 class_name: 

42 classname for the model class 

43 

44 Returns: 

45 None 

46 

47 Examples:: 

48 

49 { 

50 "type": "object", 

51 "properties": { 

52 "number": { 

53 "type": "number" 

54 }, 

55 "street_name": { 

56 "type": "string" 

57 }, 

58 "street_type": { 

59 "type": "string", 

60 "enum": ["Street", "Avenue", "Boulevard"] 

61 } 

62 } 

63 } 

64 """ 

65 if isinstance(path, str): 

66 path = Path(path) 

67 path.parent.mkdir(parents=True, exist_ok=True) 

68 

69 if isinstance(schema_type, str): 

70 schema_type = InputFileType(schema_type) 

71 

72 with TemporaryDirectory() as temp: 

73 temp = Path(temp) 

74 output = Path(temp).joinpath(f"{uuid4()}.py") 

75 

76 if url: 

77 schema = parse.urlparse(url) 

78 if not schema: 

79 raise ValueError( 

80 "Missing argument! Either 'url' or 'schema' " "must be provided" 

81 ) 

82 

83 generate( 

84 input_=schema, 

85 input_file_type=schema_type, 

86 output=output, 

87 class_name=class_name, 

88 ) 

89 

90 # move temporary file to output directory 

91 shutil.move(str(output), str(path)) 

92 

93 

94def create_context_entity_model( 

95 name: str = None, 

96 data: Dict = None, 

97 validators: Dict[str, Any] = None, 

98 path: Union[Path, str] = None, 

99) -> Type["ContextEntity"]: 

100 r""" 

101 Creates a ContextEntity-Model from a dict: 

102 

103 Args: 

104 name: 

105 name of the model 

106 data: 

107 dictionary containing the data structure 

108 validators (optional): 

109 validators for the new model 

110 path: 

111 if present the model will written to *.py file if file ending 

112 *.py is used and to json-schema if *.json is used. 

113 

114 Example: 

115 

116 >>> def username_alphanumeric(cls, value): 

117 assert v.value.isalnum(), 'must be numeric' 

118 return value 

119 

120 >>> model = create_context_entity_model( 

121 name='MyModel', 

122 data={ 

123 'id': 'MyId', 

124 'type':'MyType', 

125 'temp': 'MyProperty'} 

126 {'validate_test': validator('temperature')( 

127 username_alphanumeric)}) 

128 

129 Returns: 

130 ContextEntity 

131 

132 """ 

133 properties = { 

134 key: (ContextAttribute, ...) 

135 for key in data.keys() 

136 if key not in ContextEntity.model_fields 

137 } 

138 model = create_model( 

139 __model_name=name or "GeneratedContextEntity", 

140 __base__=ContextEntity, 

141 __validators__=validators or {}, 

142 **properties, 

143 ) 

144 

145 # if path exits a file will be generated that contains the model 

146 if path: 

147 if isinstance(path, str): 

148 path = Path(path) 

149 

150 with TemporaryDirectory() as temp: 

151 temp = Path(temp) 

152 output = Path(temp).joinpath(f"{uuid4()}.json") 

153 output.touch(exist_ok=True) 

154 with output.open("w") as f: 

155 json.dump(model.model_json_schema(), f, indent=2) 

156 if path.suffix == ".json": 

157 # move temporary file to output directory 

158 shutil.move(str(output), str(path)) 

159 elif path.suffix == ".py": 

160 create_data_model_file(path=path, schema=output, class_name=name) 

161 return model