| 126 | |
| 127 | |
| 128 | class SchemaGenerator(BaseSchemaGenerator): |
| 129 | def __init__(self, base_schema: dict[str, Any]) -> None: |
| 130 | self.base_schema = base_schema |
| 131 | |
| 132 | def get_schema(self, routes: list[BaseRoute]) -> dict[str, Any]: |
| 133 | schema = dict(self.base_schema) |
| 134 | schema.setdefault("paths", {}) |
| 135 | openapi_version_match = re.match(r"^(\d+)\.(\d+)", str(schema.get("openapi", ""))) |
| 136 | supports_query = openapi_version_match is not None and tuple(map(int, openapi_version_match.groups())) >= (3, 2) |
| 137 | endpoints_info = self.get_endpoints(routes) |
| 138 | |
| 139 | for endpoint in endpoints_info: |
| 140 | if endpoint.http_method == "query" and not supports_query: |
| 141 | continue |
| 142 | parsed = self.parse_docstring(endpoint.func) |
| 143 | |
| 144 | if not parsed: |
| 145 | continue |
| 146 | |
| 147 | if endpoint.path not in schema["paths"]: |
| 148 | schema["paths"][endpoint.path] = {} |
| 149 | |
| 150 | schema["paths"][endpoint.path][endpoint.http_method] = parsed |
| 151 | |
| 152 | return schema |
no outgoing calls