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
« 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
7The ODU CS 411W Team Red thanks Sean Baker for finding the OpenAI example.
8"""
10from jsonref import JsonRef, loads # pylint: disable=import-error
12from common.models.openapi_spec import OpenAPISpec
15class OpenAPISchemaAdapter:
16 """Adapter to OpenAPI schema from the spec text to the Pydantic dictionary
17 format used in CueCode algorithms"""
19 def __init__(self, spec: OpenAPISpec):
20 self.spec = spec
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)
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.
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
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
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