Improve OpenAPI scripts

scripts/openapi-update-classes.sh

@@ -210,12 +210,12 @@ update() {
   commit "Sort attributes and methods in $class" && unchanged "sort" || changed "sort" "sort" || return 0
 
   # apply schemas to class
-  ("$python" "$openapi" apply "$spec" "$index" "${classes[@]}" && echo) 1>&2 || failed "$class" || return 0
+  ("$python" "$openapi" apply "$source_path" "$spec" "$index" "${classes[@]}" && echo) 1>&2 || failed "$class" || return 0
   commit "Updated $class according to API spec" && unchanged "$class" || changed "$class" "$class" || return 0
 
   # apply schemas to test class
   if [ ${#classes_with_tests[@]} -gt 0 ]; then
-    ("$python" "$openapi" apply --tests "$spec" "$index" "${classes_with_tests[@]}" && echo) 1>&2 || failed "tests" || return 0
+    ("$python" "$openapi" apply --tests "$source_path" "$spec" "$index" "${classes_with_tests[@]}" && echo) 1>&2 || failed "tests" || return 0
   fi
   # do not perform linting as part of the commit as this step
   # introduces imports that might be needed by assertions

scripts/openapi.py

@@ -2145,22 +2145,32 @@ def index(self, github_path: str, spec_file: str, index_filename: str, check_ver
                 class_to_descendants[cls].append(name)
         class_to_descendants = {cls: sorted(descendants) for cls, descendants in class_to_descendants.items()}
 
+        path_to_call_methods = {}
         path_to_return_classes = {}
         schema_to_classes = {}
-        for name, cls in classes.items():
+        for cls_name, cls in classes.items():
             # construct path-to-class index
-            for method in cls.get("methods", {}).values():
+            for method_name, method in cls.get("methods", {}).items():
                 path = method.get("call", {}).get("path")
                 if not path:
                     continue
                 verb = method.get("call", {}).get("verb", "").lower()
                 if not verb:
                     if self.verbose:
-                        print(f"Unknown verb for path {path} of class {name}")
+                        print(f"Unknown verb for path {path} of class {cls_name}")
                     continue
 
                 if not path.startswith("/") and self.verbose:
                     print(f"Unsupported path: {path}")
+
+                if path not in path_to_call_methods:
+                    path_to_call_methods[path] = {}
+                if verb not in path_to_call_methods[path]:
+                    path_to_call_methods[path][verb] = set()
+                path_to_call_methods[path][verb] = sorted(
+                    list(set(path_to_call_methods[path][verb]).union({".".join([cls_name, method_name])}))
+                )
+
                 returns = method.get("returns", [])
                 if path not in path_to_return_classes:
                     path_to_return_classes[path] = {}
@@ -2174,7 +2184,7 @@ def index(self, github_path: str, spec_file: str, index_filename: str, check_ver
             for schema in cls.get("schemas"):
                 if schema not in schema_to_classes:
                     schema_to_classes[schema] = []
-                schema_to_classes[schema].append(name)
+                schema_to_classes[schema].append(cls_name)
 
         print(f"Indexed {len(classes)} classes")
         print(f"Indexed {len(path_to_return_classes)} paths")
@@ -2209,6 +2219,7 @@ def index(self, github_path: str, spec_file: str, index_filename: str, check_ver
             "paths": paths,
             "indices": {
                 "class_to_descendants": class_to_descendants,
+                "path_to_call_methods": path_to_call_methods,
                 "path_to_return_classes": path_to_return_classes,
                 "schema_to_classes": schema_to_classes,
                 "return_schema_to_paths": return_schema_to_paths,
@@ -2399,6 +2410,11 @@ def suggest_schemas(
         with open(index_filename) as r:
             index = json.load(r)
 
+        classes = index.get("classes", {})
+        schema_to_classes = index.get("indices", {}).get("schema_to_classes", {})
+        path_to_call_methods = index.get("indices", {}).get("path_to_call_methods", {})
+        path_to_return_classes = index.get("indices", {}).get("path_to_return_classes", {})
+
         schemas_added = 0
         schemas_suggested = 0
 
@@ -2436,6 +2452,7 @@ def inner_return_type(return_type: str) -> list[str]:
 
         # suggest schemas based on properties of classes
         available_schemas = {}
+        schema_returned_by = defaultdict(set)
         unimplemented_schemas = set()
         for cls in self.classes.values():
             schemas: list[str] = cls.get("schemas", [])
@@ -2453,38 +2470,46 @@ def inner_return_type(return_type: str) -> list[str]:
                 for property_name, property_spec_type in schema.get("properties", {}).items():
                     property = properties.get(property_name, {})
                     returns = property.get("returns", [])
-                    for ret in returns:
-                        cls_names = set(inner_return_type(ret))
-                        for cls_name in cls_names:
-                            if class_names and cls_name not in class_names:
-                                continue
-                            if cls_name in ["bool", "int", "str", "datetime", "list", "dict", "Any"]:
-                                continue
-                            if cls_name not in available_schemas:
-                                available_schemas[cls_name] = {}
-                            key = (cls.get("name"), property_name)
-                            if key not in available_schemas[cls_name]:
-                                available_schemas[cls_name][key] = []
-                            spec_type = self.get_inner_spec_types(
-                                property_spec_type, schema_path + ["properties", property_name]
-                            )
-                            for st in spec_type:
-                                if st.lstrip("#") not in index.get("indices", {}).get("schema_to_classes", {}):
-                                    unimplemented_schemas.add(st.lstrip("#"))
-                            available_schemas[cls_name][key].extend(spec_type)
+                    cls_names = {ir for ret in returns for ir in inner_return_type(ret)}
+                    spec_type = self.get_inner_spec_types(
+                        property_spec_type, schema_path + ["properties", property_name]
+                    )
+
+                    for cls_name in cls_names:
+                        if class_names and cls_name not in class_names:
+                            continue
+                        if cls_name in ["bool", "int", "str", "datetime", "list", "dict", "Any"]:
+                            continue
+                        if cls_name not in available_schemas:
+                            available_schemas[cls_name] = {}
+                        key = (cls.get("name"), property_name)
+                        if key not in available_schemas[cls_name]:
+                            available_schemas[cls_name][key] = []
+                        for st in spec_type:
+                            st = st.lstrip("#")
+                            schema_returned_by[st].add(key)
+                            if st not in schema_to_classes:
+                                unimplemented_schemas.add(st)
+                            # only add as available schema for cls_name if this schema
+                            # is not implemented by any other class in the union (cls_names)
+                            if not any(
+                                st in classes.get(n, {}).get("schemas", []) for n in cls_names.difference(set(cls_name))
+                            ):
+                                available_schemas[cls_name][key].append(st)
 
         for schema in sorted(list(unimplemented_schemas)):
             print(f"schema not implemented: {schema}")
 
-        classes = index.get("classes", {})
-
         for cls, provided_schemas in sorted(available_schemas.items()):
             available = set()
             implemented = classes.get(cls, {}).get("schemas", [])
             providing_properties = []
 
             for providing_property, spec_types in provided_schemas.items():
-                available = available.union([t.lstrip("#") for t in spec_types])
+                for spec_type in spec_types:
+                    spec_type = spec_type.lstrip("#")
+                    available.add(spec_type)
+
                 providing_properties.append(providing_property)
 
             schemas_to_implement = sorted(list(available.difference(set(implemented))))
@@ -2494,6 +2519,10 @@ def inner_return_type(return_type: str) -> list[str]:
                 print(f"Class {cls}:")
                 for schema_to_implement in sorted(schemas_to_implement):
                     print(f"- should implement schema {schema_to_implement}")
+                    if schema_returned_by[schema_to_implement]:
+                        print("  Properties returning the schema:")
+                        for providing_class, providing_property in sorted(schema_returned_by[schema_to_implement]):
+                            print(f"  - {providing_class}.{providing_property}")
                 # for schema_to_remove in sorted(schemas_to_remove):
                 #    print(f"- should not implement schema {schema_to_remove}")
                 print("Properties returning the class:")
@@ -2515,10 +2544,9 @@ def inner_return_type(return_type: str) -> list[str]:
 
         # suggest schemas based on API calls
         available_schemas = {}
+        schema_returned_by = defaultdict(set)
         ignored_schemas = set(index.get("config", {}).get("ignored_schemas", {}))
-        paths = set(spec.get("paths", {}).keys()).union(
-            index.get("indices", {}).get("path_to_return_classes", {}).keys()
-        )
+        paths = set(spec.get("paths", {}).keys()).union(path_to_return_classes.keys())
         for path in paths:
             for verb in spec.get("paths", {}).get(path, {}).keys():
                 responses_of_path = spec.get("paths", {}).get(path, {}).get(verb, {}).get("responses", {})
@@ -2547,6 +2575,8 @@ def inner_return_type(return_type: str) -> list[str]:
                         if verb not in available_schemas[cls]:
                             available_schemas[cls][verb] = {}
                         available_schemas[cls][verb][path] = set(schemas_of_path)
+                        for schema in schemas_of_path:
+                            schema_returned_by[schema].add((verb, path))
 
         for cls, available_verbs in sorted(available_schemas.items(), key=lambda v: v[0]):
             if cls in ["bool", "str", "None"]:
@@ -2575,6 +2605,15 @@ def inner_return_type(return_type: str) -> list[str]:
                 print(f"Class {cls}:")
                 for schema_to_implement in sorted(schemas_to_implement):
                     print(f"- should implement schema {schema_to_implement}")
+                    methods = set()
+                    for verb, verb_paths in paths.items():
+                        for path in verb_paths:
+                            if (verb, path) in schema_returned_by[schema_to_implement]:
+                                methods = methods.union(path_to_call_methods.get(path, {}).get(verb, set()))
+                    if methods:
+                        print("  Methods returning the schema:")
+                        for method in methods:
+                            print(f"  - {method}")
                 # for schema_to_remove in sorted(schemas_to_remove):
                 #    print(f"- should not implement schema {schema_to_remove}")
                 print("Paths returning the class:")