|
| 1 | +'''Contsructor to take a Python dict containing an API Documentation and |
| 2 | +create a HydraDoc object for it''' |
| 3 | +import re |
| 4 | +import json |
| 5 | +from hydrus.samples.doc_writer_sample import api_doc as sample_document |
| 6 | +from hydrus.hydraspec.doc_writer import HydraDoc, HydraClass, HydraClassProp, HydraClassOp |
| 7 | +from hydrus.hydraspec.doc_writer import HydraStatus |
| 8 | +from typing import Any, Dict, Match, Optional, Tuple, Union |
| 9 | + |
| 10 | + |
| 11 | +def error_mapping(body: str = None) -> str: |
| 12 | + """Function returns starting error message based on its body type. |
| 13 | + :param body: Params type for error message |
| 14 | + :return string: Error message for input key |
| 15 | + """ |
| 16 | + error_map = { |
| 17 | + "doc": "The API Documentation must have", |
| 18 | + "class_dict": "Class must have", |
| 19 | + "supported_prop": "Property must have", |
| 20 | + "supported_op": "Operation must have", |
| 21 | + "possible_status": "Status must have" |
| 22 | + } |
| 23 | + return error_map[body] |
| 24 | + |
| 25 | + |
| 26 | +def input_key_check( |
| 27 | + body: Dict[str, Any], key: str = None, |
| 28 | + body_type: str = None, literal: bool = False) -> dict: |
| 29 | + """Function to validate key inside the dictonary payload |
| 30 | + :param body: JSON body in which we have to check the key |
| 31 | + :param key: To check if its value exit in the body |
| 32 | + :param body_type: Name of JSON body |
| 33 | + :param literal: To check whether we need to convert the value |
| 34 | + :return string: Value of the body |
| 35 | +
|
| 36 | + Raises: |
| 37 | + SyntaxError: If the `body` does not include any entry for `key`. |
| 38 | +
|
| 39 | + """ |
| 40 | + try: |
| 41 | + if literal: |
| 42 | + return convert_literal(body[key]) |
| 43 | + return body[key] |
| 44 | + except KeyError: |
| 45 | + raise SyntaxError("{0} [{1}]".format(error_mapping(body_type), key)) |
| 46 | + |
| 47 | + |
| 48 | +def create_doc(doc: Dict[str, Any], HYDRUS_SERVER_URL: str = None, |
| 49 | + API_NAME: str = None) -> HydraDoc: |
| 50 | + """Create the HydraDoc object from the API Documentation. |
| 51 | +
|
| 52 | + Raises: |
| 53 | + SyntaxError: If the `doc` doesn't have an entry for `@id` key. |
| 54 | + SyntaxError: If the `@id` key of the `doc` is not of |
| 55 | + the form : '[protocol] :// [base url] / [entrypoint] / vocab' |
| 56 | +
|
| 57 | + """ |
| 58 | + # Check @id |
| 59 | + try: |
| 60 | + id_ = doc["@id"] |
| 61 | + except KeyError: |
| 62 | + raise SyntaxError("The API Documentation must have [@id]") |
| 63 | + |
| 64 | + # Extract base_url, entrypoint and API name |
| 65 | + match_obj = re.match(r'(.*)://(.*)/(.*)/vocab#?', id_, re.M | re.I) |
| 66 | + if match_obj: |
| 67 | + base_url = "{0}://{1}/".format(match_obj.group(1), match_obj.group(2)) |
| 68 | + entrypoint = match_obj.group(3) |
| 69 | + |
| 70 | + # Syntax checks |
| 71 | + else: |
| 72 | + raise SyntaxError( |
| 73 | + "The '@id' of the Documentation must be of the form:\n" |
| 74 | + "'[protocol] :// [base url] / [entrypoint] / vocab'") |
| 75 | + doc_keys = { |
| 76 | + "description": False, |
| 77 | + "title": False, |
| 78 | + "supportedClass": False, |
| 79 | + "@context": False, |
| 80 | + "possibleStatus": False |
| 81 | + } |
| 82 | + result = {} |
| 83 | + for k, literal in doc_keys.items(): |
| 84 | + result[k] = input_key_check(doc, k, "doc", literal) |
| 85 | + |
| 86 | + # EntryPoint object |
| 87 | + # getEntrypoint checks if all classes have @id |
| 88 | + entrypoint_obj = get_entrypoint(doc) |
| 89 | + |
| 90 | + # Main doc object |
| 91 | + if HYDRUS_SERVER_URL is not None and API_NAME is not None: |
| 92 | + apidoc = HydraDoc( |
| 93 | + API_NAME, result["title"], result["description"], API_NAME, HYDRUS_SERVER_URL) |
| 94 | + else: |
| 95 | + apidoc = HydraDoc( |
| 96 | + entrypoint, result["title"], result["description"], entrypoint, base_url) |
| 97 | + |
| 98 | + # additional context entries |
| 99 | + for entry in result["@context"]: |
| 100 | + apidoc.add_to_context(entry, result["@context"][entry]) |
| 101 | + |
| 102 | + # add all parsed_classes |
| 103 | + for class_ in result["supportedClass"]: |
| 104 | + class_obj, collection, collection_path = create_class( |
| 105 | + entrypoint_obj, class_) |
| 106 | + if class_obj: |
| 107 | + apidoc.add_supported_class( |
| 108 | + class_obj, collection=collection, collection_path=collection_path) |
| 109 | + |
| 110 | + # add possibleStatus |
| 111 | + for status in result["possibleStatus"]: |
| 112 | + status_obj = create_status(status) |
| 113 | + apidoc.add_possible_status(status_obj) |
| 114 | + |
| 115 | + apidoc.add_baseResource() |
| 116 | + apidoc.add_baseCollection() |
| 117 | + apidoc.gen_EntryPoint() |
| 118 | + return apidoc |
| 119 | + |
| 120 | + |
| 121 | +def create_class( |
| 122 | + entrypoint: Dict[str, Any], |
| 123 | + class_dict: Dict[str, Any]) -> Tuple[HydraClass, bool, str]: |
| 124 | + """Create HydraClass objects for classes in the API Documentation.""" |
| 125 | + # Base classes not used |
| 126 | + exclude_list = ['http://www.w3.org/ns/hydra/core#Resource', |
| 127 | + 'http://www.w3.org/ns/hydra/core#Collection', |
| 128 | + entrypoint["@id"]] |
| 129 | + id_ = class_dict["@id"] |
| 130 | + if id_ in exclude_list: |
| 131 | + return None, None, None |
| 132 | + match_obj = re.match(r'vocab:(.*)', id_, re.M | re.I) |
| 133 | + if match_obj: |
| 134 | + id_ = match_obj.group(1) |
| 135 | + |
| 136 | + doc_keys = { |
| 137 | + "supportedProperty": False, |
| 138 | + "title": False, |
| 139 | + "description": False, |
| 140 | + "supportedOperation": False |
| 141 | + } |
| 142 | + |
| 143 | + result = {} |
| 144 | + for k, literal in doc_keys.items(): |
| 145 | + result[k] = input_key_check(class_dict, k, "class_dict", literal) |
| 146 | + |
| 147 | + # See if class_dict is a Collection Class |
| 148 | + # type: Union[Match[Any], bool] |
| 149 | + collection = re.match(r'(.*)Collection(.*)', result["title"], re.M | re.I) |
| 150 | + if collection: |
| 151 | + return None, None, None |
| 152 | + |
| 153 | + # Check if class has it's own endpoint |
| 154 | + endpoint, path = class_in_endpoint(class_dict, entrypoint) |
| 155 | + |
| 156 | + # Check if class has a Collection |
| 157 | + collection, collection_path = collection_in_endpoint( |
| 158 | + class_dict, entrypoint) |
| 159 | + |
| 160 | + # Create the HydraClass object |
| 161 | + class_ = HydraClass( |
| 162 | + id_, result["title"], result["description"], path, endpoint=endpoint) |
| 163 | + |
| 164 | + # Add supportedProperty for the Class |
| 165 | + for prop in result["supportedProperty"]: |
| 166 | + prop_obj = create_property(prop) |
| 167 | + class_.add_supported_prop(prop_obj) |
| 168 | + |
| 169 | + # Add supportedOperation for the Class |
| 170 | + for op in result["supportedOperation"]: |
| 171 | + op_obj = create_operation(op) |
| 172 | + class_.add_supported_op(op_obj) |
| 173 | + |
| 174 | + return class_, collection, collection_path |
| 175 | + |
| 176 | + |
| 177 | +def get_entrypoint(doc: Dict[str, Any]) -> Dict[str, Any]: |
| 178 | + """Find and return the entrypoint object in the doc. |
| 179 | +
|
| 180 | + Raises: |
| 181 | + SyntaxError: If any supportedClass in the API Documentation does |
| 182 | + not have an `@id` key. |
| 183 | + SyntaxError: If no EntryPoint is found when searching in the Api Documentation. |
| 184 | +
|
| 185 | + """ |
| 186 | + |
| 187 | + # Search supportedClass |
| 188 | + for class_ in doc["supportedClass"]: |
| 189 | + # Check the @id for each class |
| 190 | + try: |
| 191 | + class_id = class_["@id"] |
| 192 | + except KeyError: |
| 193 | + raise SyntaxError("Each supportedClass must have [@id]") |
| 194 | + # Match with regular expression |
| 195 | + match_obj = re.match(r'vocab:(.*)EntryPoint', class_id) |
| 196 | + # Return the entrypoint object |
| 197 | + if match_obj: |
| 198 | + return class_ |
| 199 | + # If not found, raise error |
| 200 | + raise SyntaxError("No EntryPoint class found") |
| 201 | + |
| 202 | + |
| 203 | +def convert_literal(literal: Any) -> Optional[Union[bool, str]]: |
| 204 | + """Convert JSON literals to Python ones. |
| 205 | +
|
| 206 | + Raises: |
| 207 | + TypeError: If `literal` is not a boolean value, a string or None. |
| 208 | +
|
| 209 | + """ |
| 210 | + |
| 211 | + # Map for the literals |
| 212 | + map_ = { |
| 213 | + "true": True, |
| 214 | + "false": False, |
| 215 | + "null": None |
| 216 | + } |
| 217 | + # Check if literal is in string format |
| 218 | + if isinstance(literal, str): |
| 219 | + # Check if the literal is valid |
| 220 | + if literal in map_: |
| 221 | + return map_[literal] |
| 222 | + return literal |
| 223 | + elif isinstance(literal, (bool,)) or literal is None: |
| 224 | + return literal |
| 225 | + else: |
| 226 | + # Raise error for non string objects |
| 227 | + raise TypeError("Literal not recognised") |
| 228 | + |
| 229 | + |
| 230 | +def create_property(supported_prop: Dict[str, Any]) -> HydraClassProp: |
| 231 | + """Create a HydraClassProp object from the supportedProperty.""" |
| 232 | + # Syntax checks |
| 233 | + |
| 234 | + doc_keys = { |
| 235 | + "property": False, |
| 236 | + "title": False, |
| 237 | + "readonly": True, |
| 238 | + "writeonly": True, |
| 239 | + "required": True |
| 240 | + } |
| 241 | + result = {} |
| 242 | + for k, literal in doc_keys.items(): |
| 243 | + result[k] = input_key_check( |
| 244 | + supported_prop, k, "supported_prop", literal) |
| 245 | + # Create the HydraClassProp object |
| 246 | + prop = HydraClassProp(result["property"], result["title"], required=result["required"], |
| 247 | + read=result["readonly"], write=result["writeonly"]) |
| 248 | + return prop |
| 249 | + |
| 250 | + |
| 251 | +def class_in_endpoint( |
| 252 | + class_: Dict[str, Any], entrypoint: Dict[str, Any]) -> Tuple[bool, bool]: |
| 253 | + """Check if a given class is in the EntryPoint object as a class. |
| 254 | +
|
| 255 | + Raises: |
| 256 | + SyntaxError: If the `entrypoint` dictionary does not include the key |
| 257 | + `supportedProperty`. |
| 258 | + SyntaxError: If any dictionary in `supportedProperty` list does not include |
| 259 | + the key `property`. |
| 260 | + SyntaxError: If any property dictionary does not include the key `label`. |
| 261 | +
|
| 262 | + """ |
| 263 | + # Check supportedProperty for the EntryPoint |
| 264 | + try: |
| 265 | + supported_property = entrypoint["supportedProperty"] |
| 266 | + except KeyError: |
| 267 | + raise SyntaxError("EntryPoint must have [supportedProperty]") |
| 268 | + |
| 269 | + # Check all endpoints in supportedProperty |
| 270 | + for prop in supported_property: |
| 271 | + # Syntax checks |
| 272 | + try: |
| 273 | + property_ = prop["property"] |
| 274 | + except KeyError: |
| 275 | + raise SyntaxError("supportedProperty must have [property]") |
| 276 | + try: |
| 277 | + label = property_["label"] |
| 278 | + except KeyError: |
| 279 | + raise SyntaxError("property must have [label]") |
| 280 | + # Match the title with regular expression |
| 281 | + |
| 282 | + if label == class_['title']: |
| 283 | + path = "/".join(property_['@id'].split("/")[1:]) |
| 284 | + return True, path |
| 285 | + return False, None |
| 286 | + |
| 287 | + |
| 288 | +def collection_in_endpoint( |
| 289 | + class_: Dict[str, Any], entrypoint: Dict[str, Any]) -> Tuple[bool, bool]: |
| 290 | + """Check if a given class is in the EntryPoint object as a collection. |
| 291 | +
|
| 292 | + Raises: |
| 293 | + SyntaxError: If the `entrypoint` dictionary does not include the key |
| 294 | + `supportedProperty`. |
| 295 | + SyntaxError: If any dictionary in `supportedProperty` list does not include |
| 296 | + the key `property`. |
| 297 | + SyntaxError: If any property dictionary does not include the key `label`. |
| 298 | +
|
| 299 | + """ |
| 300 | + # Check supportedProperty for the EntryPoint |
| 301 | + try: |
| 302 | + supported_property = entrypoint["supportedProperty"] |
| 303 | + except KeyError: |
| 304 | + raise SyntaxError("EntryPoint must have [supportedProperty]") |
| 305 | + |
| 306 | + # Check all endpoints in supportedProperty |
| 307 | + for prop in supported_property: |
| 308 | + # Syntax checks |
| 309 | + try: |
| 310 | + property_ = prop["property"] |
| 311 | + except KeyError: |
| 312 | + raise SyntaxError("supportedProperty must have [property]") |
| 313 | + try: |
| 314 | + label = property_["label"] |
| 315 | + except KeyError: |
| 316 | + raise SyntaxError("property must have [label]") |
| 317 | + # Match the title with regular expression |
| 318 | + if label == "{}Collection".format(class_["title"]): |
| 319 | + path = "/".join(property_['@id'].split("/")[1:]) |
| 320 | + return True, path |
| 321 | + return False, None |
| 322 | + |
| 323 | + |
| 324 | +def create_operation(supported_op: Dict[str, Any]) -> HydraClassOp: |
| 325 | + """Create a HyraClassOp object from the supportedOperation.""" |
| 326 | + # Syntax checks |
| 327 | + doc_keys = { |
| 328 | + "title": False, |
| 329 | + "method": False, |
| 330 | + "expects": True, |
| 331 | + "returns": True, |
| 332 | + "possibleStatus": False |
| 333 | + } |
| 334 | + result = {} |
| 335 | + for k, literal in doc_keys.items(): |
| 336 | + result[k] = input_key_check(supported_op, k, "supported_op", literal) |
| 337 | + |
| 338 | + # Create the HydraClassOp object |
| 339 | + op_ = HydraClassOp(result["title"], result["method"], |
| 340 | + result["expects"], result["returns"], result["possibleStatus"]) |
| 341 | + return op_ |
| 342 | + |
| 343 | + |
| 344 | +def create_status(possible_status: Dict[str, Any]) -> HydraStatus: |
| 345 | + """Create a HydraStatus object from the possibleStatus.""" |
| 346 | + # Syntax checks |
| 347 | + doc_keys = { |
| 348 | + "title": False, |
| 349 | + "statusCode": False, |
| 350 | + "description": True |
| 351 | + } |
| 352 | + result = {} |
| 353 | + for k, literal in doc_keys.items(): |
| 354 | + result[k] = input_key_check( |
| 355 | + possible_status, k, "possible_status", literal) |
| 356 | + # Create the HydraStatus object |
| 357 | + status = HydraStatus(result["statusCode"], |
| 358 | + result["title"], result["description"]) |
| 359 | + return status |
| 360 | + |
| 361 | + |
| 362 | +if __name__ == "__main__": |
| 363 | + api_doc = create_doc(sample_document.generate()) |
| 364 | + print(json.dumps(api_doc.generate(), indent=4, sort_keys=True)) |
0 commit comments