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

1""" 

2Code generator for data models from schema.json descriptions 

3""" 

4import json 

5import shutil 

6from pathlib import Path 

7from tempfile import TemporaryDirectory 

8from typing import Union, Dict, Any, Type 

9from urllib import parse 

10from uuid import uuid4 

11from datamodel_code_generator import InputFileType, generate, ParseResult 

12from pydantic import create_model 

13 

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

15 

16 

17def create_data_model_file(*, 

18 path: Union[Path, str], 

19 url: str = None, 

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

21 schema_type: Union[str, InputFileType] = 

22 InputFileType.JsonSchema, 

23 class_name: str = None 

24 ) -> None: 

25 """ 

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

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

28 types are defined but the underlying toolbox. 

29 

30 Many data models suited for FIWARE are located here: 

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

32 

33 Args: 

34 path: 

35 path where the generated code should saved 

36 url: 

37 url to download the definition from 

38 schema_type (str): 

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

40 class_name: 

41 classname for the model class 

42 

43 Returns: 

44 None 

45 

46 Examples:: 

47 

48 { 

49 "type": "object", 

50 "properties": { 

51 "number": { 

52 "type": "number" 

53 }, 

54 "street_name": { 

55 "type": "string" 

56 }, 

57 "street_type": { 

58 "type": "string", 

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

60 } 

61 } 

62 } 

63 """ 

64 if isinstance(path, str): 

65 path = Path(path) 

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

67 

68 if isinstance(schema_type, str): 

69 schema_type = InputFileType(schema_type) 

70 

71 with TemporaryDirectory() as temp: 

72 temp = Path(temp) 

73 output = Path(temp).joinpath(f'{uuid4()}.py') 

74 

75 if url: 

76 schema = parse.urlparse(url) 

77 if not schema: 

78 raise ValueError("Missing argument! Either 'url' or 'schema' " 

79 "must be provided") 

80 

81 generate( 

82 input_=schema, 

83 input_file_type=schema_type, 

84 output=output, 

85 class_name=class_name) 

86 

87 # move temporary file to output directory 

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

89 

90 

91def create_context_entity_model(name: str = None, 

92 data: Dict = None, 

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

94 path: Union[Path, str] = None) -> \ 

95 Type['ContextEntity']: 

96 r""" 

97 Creates a ContextEntity-Model from a dict: 

98 

99 Args: 

100 name: 

101 name of the model 

102 data: 

103 dictionary containing the data structure 

104 validators (optional): 

105 validators for the new model 

106 path: 

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

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

109 

110 Example: 

111 

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

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

114 return value 

115 

116 >>> model = create_context_entity_model( 

117 name='MyModel', 

118 data={ 

119 'id': 'MyId', 

120 'type':'MyType', 

121 'temp': 'MyProperty'} 

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

123 username_alphanumeric)}) 

124 

125 Returns: 

126 ContextEntity 

127 

128 """ 

129 properties = {key: (ContextAttribute, ...) for key in data.keys() if 

130 key not in ContextEntity.model_fields} 

131 model = create_model( 

132 __model_name=name or 'GeneratedContextEntity', 

133 __base__=ContextEntity, 

134 __validators__=validators or {}, 

135 **properties 

136 ) 

137 

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

139 if path: 

140 if isinstance(path, str): 

141 path=Path(path) 

142 

143 with TemporaryDirectory() as temp: 

144 temp = Path(temp) 

145 output = Path(temp).joinpath(f'{uuid4()}.json') 

146 output.touch(exist_ok=True) 

147 with output.open('w') as f: 

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

149 if path.suffix == '.json': 

150 # move temporary file to output directory 

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

152 elif path.suffix == '.py': 

153 create_data_model_file(path=path, 

154 schema=output, 

155 class_name=name) 

156 return model