Coverage for src/configuration/openapi_schema_adapter.py: 41%

29 statements  

« prev     ^ index     » next       coverage.py v7.6.10, created at 2025-03-25 22:39 -0400

1"""Adapter to OpenAPI schema from the spec text to the dictionary format used 

2in CueCode algorithms. Inspiration and code examples drawn from two sources under 

3license: 

4- https://github.com/openai/openai-cookbook/blob/main/examples/Function_calling_with_an_OpenAPI_spec.ipynb # pylint: disable=line-too-long 

5- https://github.com/python-openapi/openapi-spec-validator?tab=readme-ov-file#python-package 

6 

7The ODU CS 411W Team Red thanks Sean Baker for finding the OpenAI example. 

8""" 

9 

10from jsonref import JsonRef, loads # pylint: disable=import-error 

11 

12from common.models.openapi_spec import OpenAPISpec 

13 

14 

15class OpenAPISchemaAdapter: 

16 """Adapter to OpenAPI schema from the spec text to the Pydantic dictionary 

17 format used in CueCode algorithms""" 

18 

19 def __init__(self, spec: OpenAPISpec): 

20 self.spec = spec 

21 

22 def get_raw_json_dict(self) -> JsonRef: 

23 """Get the spec as a dict, with all JSON references lazily dereferenced 

24 on access, to allow us to follow the `'$ref'`s in the OpenAPI spec""" 

25 return loads(self.spec.spec_text) 

26 

27 def get_cleaned_json_dict(self) -> JsonRef: 

28 """Get the preprocessed version of the spec as a dict, 

29 with all JSON references lazily dereferenced 

30 on access, to allow us to follow the `'$ref'`s in the OpenAPI spec. 

31 

32 Preprocessing applies fixes to common, easily corrected CueCode 

33 compatability problems in OpenAPI specs 

34 """ 

35 raw = self.get_raw_json_dict() 

36 cleaned = OpenAPISchemaAdapter._fix_empty_schemas(raw) 

37 cleaned = OpenAPISchemaAdapter._fix_broken_security(cleaned) 

38 return cleaned 

39 

40 @staticmethod 

41 def _fix_empty_schemas(d: JsonRef) -> JsonRef: 

42 for k, v in d.items(): 

43 if isinstance(v, JsonRef): 

44 OpenAPISchemaAdapter._fix_broken_security(v) 

45 elif k == "schema" and isinstance(v, list) and not v: 

46 d[k] = {} 

47 return d 

48 

49 @staticmethod 

50 def _fix_broken_security(d: JsonRef) -> JsonRef: 

51 for k, v in d.items(): 

52 if isinstance(v, JsonRef): 

53 OpenAPISchemaAdapter._fix_broken_security(v) 

54 elif k == "security" and isinstance(v, list): 

55 while {} in v: 

56 v.remove({}) 

57 return d