add validation and improve ref resolving

Author: tkfoss

README.md

@@ -34,6 +34,14 @@ pip install dist/serverless_openapi_generator-1.0.0-py3-none-any.whl
 
 Once installed, you can use the `openapi-gen` command-line tool.
 
+### Running from GitHub
+
+You can also run the tool directly from GitHub using `uvx`:
+```bash
+uvx --from git+https://github.com/tkfoss/python-serverless-openapi-documentation.git openapi-gen path/to/your/serverless.yml openapi.json
+```
+For more information on running tools with `uv`, see the [official documentation](https://docs.astral.sh/uv/guides/tools/#running-tools).
+
 **To Run:**
 ```bash
 openapi-gen path/to/your/serverless.yml openapi.json --openApiVersion 3.0.3
@@ -46,6 +54,21 @@ serverless_yml_path      Path to the serverless.yml file. (Required)
 output_file_path         Path to the output OpenAPI JSON file. (Required)
 --openApiVersion         The OpenAPI version to generate for. Default: 3.0.3
 --pre-hook               Path to a Python script to run before generation.
+--validate               Validate the generated OpenAPI spec.
+```
+
+### Validation
+
+This tool also includes a script to validate an OpenAPI specification file against the OpenAPI 3.0.3 specification.
+
+**To Run:**
+```bash
+openapi-validate path/to/your/openapi.json
+```
+
+You can also use the `--validate` flag with the `openapi-gen` command to automatically validate the generated spec:
+```bash
+openapi-gen path/to/your/serverless.yml openapi.json --validate
 ```
 
 ### Configuration

pyproject.toml

@@ -21,10 +21,12 @@ dependencies = [
     "requests",
     "referencing",
     "jsonschema-spec",
+    "openapi-spec-validator",
 ]
 
 [tool.setuptools.packages.find]
 where = ["src"]
 
 [project.scripts]
 openapi-gen = "serverless_openapi_generator.openapi_generator:main"
+openapi-validate = "openapi_spec_validator.__main__:main"

src/serverless_openapi_generator/openapi_generator.py

@@ -274,12 +274,13 @@ def create_media_type_object(self, models):
             else:
                 # Fallback for inline schemas, though we'll focus on named models
                 continue
-
-            media_type_obj[media_type] = {
-                'schema': {
-                    '$ref': f"#/components/schemas/{model_name}"
+            
+            if model_name in self.open_api['components']['schemas']:
+                media_type_obj[media_type] = {
+                    'schema': {
+                        '$ref': f"#/components/schemas/{model_name}"
+                    }
                 }
-            }
         return media_type_obj
 
     def create_request_body(self, request_body_doc):
@@ -391,6 +392,7 @@ def main():
     parser.add_argument('output_file_path', type=str, help='Path to the output OpenAPI JSON file')
     parser.add_argument('--openApiVersion', type=str, default='3.0.3', help='OpenAPI version to use')
     parser.add_argument('--pre-hook', type=str, help='Path to a Python script to run before generation')
+    parser.add_argument('--validate', action='store_true', help='Validate the generated OpenAPI spec')
     args = parser.parse_args()
 
     # Execute the pre-hook script if provided
@@ -426,5 +428,14 @@ def main():
     except IOError as e:
         print(f"Error writing to output file: {e}")
 
+    if args.validate:
+        from openapi_spec_validator import validate
+        print("Validating generated spec...")
+        try:
+            validate(open_api_spec)
+            print("Validation successful.")
+        except Exception as e:
+            print(f"Validation failed: {e}")
+
 if __name__ == '__main__':
     main()

src/serverless_openapi_generator/schema_handler.py

@@ -87,6 +87,67 @@ def create_schema(self, name, schema_definition):
         self.open_api['components']['schemas'][name] = final_schema
         return f"#/components/schemas/{name}"
 
+    def _clean_schema(self, schema):
+        if isinstance(schema, dict):
+            # Replace $defs with definitions and update refs
+            if '$defs' in schema:
+                if 'definitions' not in schema:
+                    schema['definitions'] = {}
+                schema['definitions'].update(schema.pop('$defs'))
+                
+                def update_refs(node):
+                    if isinstance(node, dict):
+                        if '$ref' in node and node['$ref'].startswith('#/$defs/'):
+                            node['$ref'] = node['$ref'].replace('#/$defs/', '#/definitions/')
+                        for key, value in node.items():
+                            update_refs(value)
+                    elif isinstance(node, list):
+                        for item in node:
+                            update_refs(item)
+                update_refs(schema)
+
+            # Replace const with enum
+            if 'const' in schema:
+                schema['enum'] = [schema.pop('const')]
+
+            # Remove propertyNames
+            if 'propertyNames' in schema:
+                del schema['propertyNames']
+            
+            # Handle nullable
+            if 'anyOf' in schema:
+                is_nullable = False
+                new_any_of = []
+                for item in schema['anyOf']:
+                    if isinstance(item, dict) and item.get('type') == 'null':
+                        is_nullable = True
+                    else:
+                        new_any_of.append(item)
+                
+                if is_nullable:
+                    for item in new_any_of:
+                        if isinstance(item, dict):
+                            item['nullable'] = True
+                
+                if not new_any_of:
+                    del schema['anyOf']
+                    schema['nullable'] = True
+                elif len(new_any_of) == 1:
+                    for key, value in new_any_of[0].items():
+                        if key not in schema:
+                            schema[key] = value
+                    del schema['anyOf']
+                    if is_nullable:
+                        schema['nullable'] = True
+                else:
+                    schema['anyOf'] = new_any_of
+
+            # Recurse
+            return {k: self._clean_schema(v) for k, v in schema.items()}
+        elif isinstance(schema, list):
+            return [self._clean_schema(item) for item in schema]
+        return schema
+
     def _resolve_schema_references(self, schema):
         schema = self._resolve_file_references(schema)
         if not isinstance(schema, dict):
@@ -100,10 +161,12 @@ def _resolve_schema_references(self, schema):
             else:
                 return schema
 
+        schema = self._clean_schema(schema)
+
         registry = Registry()
         if "definitions" in schema:
             for name, sub_schema in schema["definitions"].items():
-                resource = Resource.from_contents(sub_schema, default_specification=DRAFT4)
+                resource = Resource.from_contents(self._clean_schema(sub_schema), default_specification=DRAFT4)
                 registry = registry.with_resource(f"#/definitions/{name}", resource)
 
         main_resource = Resource.from_contents(schema, default_specification=DRAFT4)
@@ -118,8 +181,11 @@ def _resolve_schema_references(self, schema):
     def _recursive_dereference(self, node, resolver):
         if isinstance(node, dict):
             if "$ref" in node:
-                resolved = resolver.lookup(node["$ref"])
-                return self._recursive_dereference(resolved.contents, resolver)
+                try:
+                    resolved = resolver.lookup(node["$ref"])
+                    return self._recursive_dereference(resolved.contents, resolver)
+                except Exception:
+                    return node
             return {k: self._recursive_dereference(v, resolver) for k, v in node.items()}
         elif isinstance(node, list):
             return [self._recursive_dereference(item, resolver) for item in node]