Method index

scripts/openapi.py:2251–2403

(
        self, github_path: str, spec_file: str | None, index_filename: str, check_verbs: bool, dry_run: bool
    )

        return True

    def index(
        self, github_path: str, spec_file: str | None, index_filename: str, check_verbs: bool, dry_run: bool
    ) -> bool:
        import multiprocessing

        config = {}
        config_file = Path(github_path) / "openapi.index.json"
        if config_file.exists():
            with config_file.open("r") as r:
                ignored_schemas = json.load(r).get("ignored OpenAPI schemas", [])
            config["ignored_schemas"] = ignored_schemas

        files = [f for f in listdir(github_path) if isfile(join(github_path, f)) and f.endswith(".py")]
        print(f"Indexing {len(files)} Python files")

        # index files in parallel
        with multiprocessing.Manager() as manager:
            classes = manager.dict()
            paths = manager.list()
            if check_verbs and not config_file.exists():
                raise RuntimeError(f"Cannot check verbs without config: {config_file}")
            indexer = IndexFileWorker(classes, config_file, paths, check_verbs)
            with multiprocessing.Pool() as pool:
                pool.map(indexer.index_file, iterable=[join(github_path, file) for file in files])
            classes = dict(classes)
            paths = merge_paths(paths)

        # construct inheritance list
        while self.extend_inheritance(classes):
            pass

        # propagate ids of base classes to derived classes
        while self.propagate_ids(classes):
            pass

        # construct class_to_descendants
        class_to_descendants = {}
        for name, descendant in classes.items():
            for cls in descendant.get("inheritance", []):
                if cls not in class_to_descendants:
                    class_to_descendants[cls] = []
                class_to_descendants[cls].append(name)
        class_to_descendants = {cls: sorted(descendants) for cls, descendants in class_to_descendants.items()}

        # add schemas of base classes to derived classes
        for name, clazz in sorted(classes.items(), key=lambda v: v[0]):
            schemas = clazz.get("schemas")
            if not schemas:
                continue

            for derived in class_to_descendants.get(name, []):
                if derived in classes:
                    derived_schemas = classes[derived].get("inherited_schemas", [])
                    for schema in schemas:
                        if schema not in derived_schemas:
                            derived_schemas.append(schema)
                    classes[derived]["inherited_schemas"] = derived_schemas