300 lines
11 KiB
Python
300 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate the browser API client from the committed OpenAPI document."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
HEADER = """/* eslint-disable */
|
|
/*
|
|
* Generated by tools/generate_api_client.py from openapi/v2.json.
|
|
* Do not edit this file directly. Run `npm --prefix frontend run api:generate`.
|
|
*/
|
|
|
|
"""
|
|
|
|
|
|
def identifier(value: str) -> str:
|
|
parts = re.split(r"[^A-Za-z0-9]+", value)
|
|
result = "".join(part[:1].upper() + part[1:] for part in parts if part)
|
|
if not result:
|
|
return "Anonymous"
|
|
return f"_{result}" if result[0].isdigit() else result
|
|
|
|
|
|
def property_name(value: str) -> str:
|
|
return value if re.match(r"^[A-Za-z_$][A-Za-z0-9_$]*$", value) else json.dumps(value)
|
|
|
|
|
|
def operation_name(operation_id: str) -> str:
|
|
if "_api_" in operation_id:
|
|
prefix = operation_id.split("_api_", 1)[0]
|
|
else:
|
|
prefix = re.sub(r"_(get|post|put|patch|delete)$", "", operation_id)
|
|
first, *rest = prefix.split("_")
|
|
prefix = first if rest and all(part == first for part in rest) else prefix
|
|
pieces = [piece for piece in prefix.split("_") if piece]
|
|
return pieces[0] + "".join(piece.capitalize() for piece in pieces[1:])
|
|
|
|
|
|
def schema_type(schema: dict[str, Any]) -> str:
|
|
if "$ref" in schema:
|
|
return f"Components['schemas'][{json.dumps(schema['$ref'].rsplit('/', 1)[-1])}]"
|
|
for key in ("anyOf", "oneOf", "allOf"):
|
|
if key in schema:
|
|
separator = " & " if key == "allOf" else " | "
|
|
return separator.join(schema_type(item) for item in schema[key])
|
|
if "enum" in schema:
|
|
return " | ".join(json.dumps(value) for value in schema["enum"])
|
|
kind = schema.get("type")
|
|
if kind == "string":
|
|
return "string"
|
|
if kind in {"integer", "number"}:
|
|
return "number"
|
|
if kind == "boolean":
|
|
return "boolean"
|
|
if kind == "null":
|
|
return "null"
|
|
if kind == "array":
|
|
return f"Array<{schema_type(schema.get('items', {}))}>"
|
|
if kind == "object" or "properties" in schema:
|
|
properties = schema.get("properties", {})
|
|
required = set(schema.get("required", []))
|
|
fields = [
|
|
f"{property_name(name)}{' ' if name in required else '?'}: {schema_type(value)}"
|
|
for name, value in properties.items()
|
|
]
|
|
additional = schema.get("additionalProperties")
|
|
if not fields:
|
|
return (
|
|
f"Record<string, {schema_type(additional)}>"
|
|
if isinstance(additional, dict)
|
|
else "Record<string, unknown>"
|
|
)
|
|
result = "{ " + "; ".join(fields) + " }"
|
|
if isinstance(additional, dict):
|
|
result += f" & Record<string, {schema_type(additional)}>"
|
|
return result
|
|
return "unknown"
|
|
|
|
|
|
def response_type(operation: dict[str, Any]) -> str:
|
|
responses = operation.get("responses", {})
|
|
successful = next((item for status, item in responses.items() if status.startswith("2")), None)
|
|
if successful is None:
|
|
return "void"
|
|
content = successful.get("content", {})
|
|
json_content = content.get("application/json")
|
|
if not json_content:
|
|
return "void"
|
|
return schema_type(json_content.get("schema", {}))
|
|
|
|
|
|
def parameter_type(parameters: list[dict[str, Any]], location: str) -> str | None:
|
|
selected = [parameter for parameter in parameters if parameter.get("in") == location]
|
|
if not selected:
|
|
return None
|
|
fields = []
|
|
for parameter in selected:
|
|
optional = "" if parameter.get("required") else "?"
|
|
parameter_schema = schema_type(parameter.get("schema", {}))
|
|
fields.append(f"{property_name(parameter['name'])}{optional}: {parameter_schema}")
|
|
return "{ " + "; ".join(fields) + " }"
|
|
|
|
|
|
def request_body_type(operation: dict[str, Any]) -> tuple[str | None, bool]:
|
|
body = operation.get("requestBody")
|
|
if not body:
|
|
return None, False
|
|
content = body.get("content", {})
|
|
json_content = content.get("application/json")
|
|
return (
|
|
schema_type(json_content.get("schema", {})) if json_content else "unknown",
|
|
bool(body.get("required")),
|
|
)
|
|
|
|
|
|
def is_event_stream(operation: dict[str, Any]) -> bool:
|
|
return any(
|
|
"text/event-stream" in response.get("content", {})
|
|
for response in operation.get("responses", {}).values()
|
|
)
|
|
|
|
|
|
def render_operation(
|
|
name: str, path: str, method: str, operation: dict[str, Any]
|
|
) -> tuple[str, str]:
|
|
parameters = operation.get("parameters", [])
|
|
path_type = parameter_type(parameters, "path")
|
|
query_type = parameter_type(parameters, "query")
|
|
body_type, body_required = request_body_type(operation)
|
|
fields: list[str] = []
|
|
if path_type:
|
|
fields.append(f"path: {path_type}")
|
|
if query_type:
|
|
fields.append(f"query?: {query_type}")
|
|
if body_type:
|
|
fields.append(f"body{' ' if body_required else '?'}: {body_type}")
|
|
result = response_type(operation)
|
|
if fields:
|
|
params_name = f"{identifier(name)}Params"
|
|
declaration = f"export type {params_name} = {{ {'; '.join(fields)} }};\n\n"
|
|
signature = f"params: {params_name}, options: RequestOptions = {{}}"
|
|
parameter_expression = "params"
|
|
else:
|
|
declaration = ""
|
|
signature = "options: RequestOptions = {}"
|
|
parameter_expression = "{}"
|
|
path_expression = json.dumps(path)
|
|
for parameter in parameters:
|
|
if parameter.get("in") == "path":
|
|
name_value = parameter["name"]
|
|
path_expression += (
|
|
f".replace({json.dumps('{' + name_value + '}')}, "
|
|
f"encodeURIComponent(String({parameter_expression}.path.{property_name(name_value)})))"
|
|
)
|
|
query_expression = ""
|
|
if query_type:
|
|
query_expression = f"\n\t\tappendQuery(url.searchParams, {parameter_expression}.query);"
|
|
body_expression = f", {parameter_expression}.body" if body_type else ""
|
|
if is_event_stream(operation):
|
|
return (
|
|
declaration,
|
|
"\t/** EventSource transport; intentionally not a JSON fetch Promise. */\n"
|
|
+ f"\t{name}Url({signature.split(', options')[0]}): URL {{\n"
|
|
+ f"\t\tconst url = new URL({path_expression}, this.baseUrl);"
|
|
+ query_expression
|
|
+ "\n\t\treturn url;\n\t}\n\n",
|
|
)
|
|
return (
|
|
declaration,
|
|
f"\tasync {name}({signature}): Promise<{result}> {{\n"
|
|
+ f"\t\tconst url = new URL({path_expression}, this.baseUrl);"
|
|
+ query_expression
|
|
+ f"\n\t\treturn request<{result}>(\n"
|
|
+ f"\t\t\turl, {json.dumps(method.upper())}, options{body_expression}\n"
|
|
+ "\t\t);\n\t}\n\n",
|
|
)
|
|
|
|
|
|
def generate(spec: dict[str, Any]) -> str:
|
|
schemas = spec.get("components", {}).get("schemas", {})
|
|
type_lines = ["export interface Components {", " schemas: {"]
|
|
for name, schema in schemas.items():
|
|
type_lines.append(f" {property_name(name)}: {schema_type(schema)};")
|
|
type_lines += [" };", "}", ""]
|
|
operation_declarations: list[str] = []
|
|
operations: list[str] = []
|
|
for path, path_item in spec.get("paths", {}).items():
|
|
for method, operation in path_item.items():
|
|
if method not in {"get", "post", "put", "patch", "delete"}:
|
|
continue
|
|
declaration, implementation = render_operation(
|
|
operation_name(operation["operationId"]), path, method, operation
|
|
)
|
|
operation_declarations.append(declaration)
|
|
operations.append(implementation)
|
|
runtime = r"""export type RequestOptions = Omit<RequestInit, "body">;
|
|
|
|
export type Problem = {
|
|
type: string;
|
|
title: string;
|
|
status: number;
|
|
detail: string;
|
|
instance: string;
|
|
code: string;
|
|
};
|
|
|
|
export class ApiError extends Error {
|
|
readonly status: number;
|
|
readonly problem?: Problem;
|
|
|
|
constructor(status: number, problem?: Problem) {
|
|
super(problem?.detail ?? `Request failed with status ${status}.`);
|
|
this.name = "ApiError";
|
|
this.status = status;
|
|
this.problem = problem;
|
|
}
|
|
}
|
|
|
|
export function isApiError(error: unknown): error is ApiError {
|
|
return error instanceof ApiError;
|
|
}
|
|
|
|
function appendQuery(search: URLSearchParams, query: Record<string, unknown> | undefined): void {
|
|
if (!query) return;
|
|
for (const [key, value] of Object.entries(query)) {
|
|
if (value === undefined || value === null) continue;
|
|
for (const item of Array.isArray(value) ? value : [value]) search.append(key, String(item));
|
|
}
|
|
}
|
|
|
|
async function request<T>(
|
|
url: URL,
|
|
method: string,
|
|
options: RequestOptions,
|
|
body?: unknown,
|
|
): Promise<T> {
|
|
const headers = new Headers(options.headers);
|
|
if (body !== undefined && !headers.has("Content-Type")) {
|
|
headers.set("Content-Type", "application/json");
|
|
}
|
|
const response = await fetch(url, {
|
|
...options,
|
|
method,
|
|
headers,
|
|
credentials: options.credentials ?? "same-origin",
|
|
body: body === undefined ? undefined : JSON.stringify(body),
|
|
});
|
|
if (response.status === 204) return undefined as T;
|
|
const contentType = response.headers.get("content-type") ?? "";
|
|
const isJson = contentType.includes("application/json")
|
|
|| contentType.includes("application/problem+json");
|
|
const payload: unknown = isJson ? await response.json() : undefined;
|
|
if (!response.ok) throw new ApiError(response.status, payload as Problem | undefined);
|
|
return payload as T;
|
|
}
|
|
|
|
export class BackupToolClient {
|
|
constructor(readonly baseUrl = window.location.origin) {}
|
|
|
|
"""
|
|
return (
|
|
HEADER
|
|
+ "\n".join(type_lines)
|
|
+ "".join(operation_declarations)
|
|
+ runtime
|
|
+ "".join(operations)
|
|
+ "}\n"
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Generate TypeScript API client from OpenAPI.")
|
|
parser.add_argument("--input", type=Path, default=Path("openapi/v2.json"))
|
|
parser.add_argument("--output", type=Path, default=Path("frontend/src/api/generated/client.ts"))
|
|
parser.add_argument("--check", action="store_true", help="fail when generated output differs")
|
|
arguments = parser.parse_args()
|
|
rendered = generate(json.loads(arguments.input.read_text(encoding="utf-8")))
|
|
if arguments.check:
|
|
if (
|
|
not arguments.output.is_file()
|
|
or arguments.output.read_text(encoding="utf-8") != rendered
|
|
):
|
|
print(f"Generated client drift: regenerate {arguments.output}")
|
|
return 1
|
|
print(f"Generated client is current: {arguments.output}")
|
|
return 0
|
|
arguments.output.parent.mkdir(parents=True, exist_ok=True)
|
|
arguments.output.write_text(rendered, encoding="utf-8")
|
|
print(f"Generated API client: {arguments.output}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|