Proxy class to generate client from OpenAPI schema.
Source code in fastagency/api/openapi/openapi.py
| def __init__(
self, servers: list[dict[str, Any]], title: Optional[str] = None, **kwargs: Any
) -> None:
"""Proxy class to generate client from OpenAPI schema."""
self._servers = servers
self._title = title
self._kwargs = kwargs
self._registered_funcs: list[Callable[..., Any]] = []
self._globals: dict[str, Any] = {}
self._security: dict[str, list[BaseSecurity]] = {}
self._security_params: dict[Optional[str], BaseSecurityParameters] = {}
|
create classmethod
Source code in fastagency/api/openapi/openapi.py
| @classmethod
def create(
cls,
*,
openapi_json: Optional[str] = None,
openapi_url: Optional[str] = None,
client_source_path: Optional[str] = None,
servers: Optional[list[dict[str, Any]]] = None,
) -> "OpenAPI":
if (openapi_json is None) == (openapi_url is None):
raise ValueError("Either openapi_json or openapi_url should be provided")
if openapi_json is None and openapi_url is not None:
with requests.get(openapi_url, timeout=10) as response:
response.raise_for_status()
openapi_json = response.text
if servers:
openapi_parsed = json.loads(openapi_json) # type: ignore [arg-type]
openapi_parsed["servers"] = servers
openapi_json = json.dumps(openapi_parsed)
with optional_temp_path(client_source_path) as td:
suffix = td.name
main_name = cls.generate_code(
input_text=openapi_json, # type: ignore [arg-type]
output_dir=td,
)
# add td to sys.path
try:
sys.path.append(str(td))
main = importlib.import_module(main_name, package=td.name) # nosemgrep
finally:
sys.path.remove(str(td))
client: OpenAPI = main.app # type: ignore [attr-defined]
client.set_globals(main, suffix=suffix)
return client
|
delete
Source code in fastagency/api/openapi/openapi.py
| def delete(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("delete", path, **kwargs)
|
generate_code classmethod
Source code in fastagency/api/openapi/openapi.py
| @classmethod
def generate_code(
cls,
input_text: str,
output_dir: Path,
disable_timestamp: bool = False,
custom_visitors: Optional[list[Path]] = None,
) -> str:
if custom_visitors is None:
custom_visitors = []
custom_visitors.append(Path(__file__).parent / "security_schema_visitor.py")
with patch_get_parameter_type():
generate_code(
input_name="openapi.json",
input_text=input_text,
encoding="utf-8",
output_dir=output_dir,
template_dir=cls._get_template_dir(),
disable_timestamp=disable_timestamp,
custom_visitors=custom_visitors,
output_model_type=DataModelType.PydanticV2BaseModel,
)
# Use unique file name for main.py
main_name = f"main_{output_dir.name}"
main_path = output_dir / f"{main_name}.py"
shutil.move(output_dir / "main.py", main_path)
# Change "from models import" to "from models_unique_name import"
with main_path.open("r") as f:
main_py_code = f.read()
main_py_code = main_py_code.replace(
"from .models import", f"from models_{output_dir.name} import"
)
with main_path.open("w") as f:
f.write(main_py_code)
# Use unique file name for models.py
models_name = f"models_{output_dir.name}"
models_path = output_dir / f"{models_name}.py"
shutil.move(output_dir / "models.py", models_path)
return main_name
|
get
Source code in fastagency/api/openapi/openapi.py
| def get(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("get", path, **kwargs)
|
get_function
Source code in fastagency/api/openapi/openapi.py
| def get_function(self, name: str) -> Callable[..., dict[str, Any]]:
for f in self._registered_funcs:
if f.__name__ == name:
return f
raise ValueError(f"Function {name} not found")
|
get_functions
Source code in fastagency/api/openapi/openapi.py
| def get_functions(self) -> list[str]:
raise DeprecationWarning(
"Use function_names property instead of get_functions method"
)
|
head
Source code in fastagency/api/openapi/openapi.py
| def head(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("head", path, **kwargs)
|
inject_parameters
inject_parameters(name: str, **kwargs: Any) -> None
Source code in fastagency/api/openapi/openapi.py
| def inject_parameters(self, name: str, **kwargs: Any) -> None:
raise NotImplementedError("Injecting parameters is not implemented yet")
|
patch
Source code in fastagency/api/openapi/openapi.py
| def patch(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("patch", path, **kwargs)
|
post
Source code in fastagency/api/openapi/openapi.py
| def post(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("post", path, **kwargs)
|
put
Source code in fastagency/api/openapi/openapi.py
| def put(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("put", path, **kwargs)
|
set_function
Source code in fastagency/api/openapi/openapi.py
| def set_function(self, name: str, func: Callable[..., dict[str, Any]]) -> None:
for i, f in enumerate(self._registered_funcs):
if f.__name__ == name:
self._registered_funcs[i] = func
return
raise ValueError(f"Function {name} not found")
|
set_globals
Source code in fastagency/api/openapi/openapi.py
| def set_globals(self, main: ModuleType, suffix: str) -> None:
xs = {k: v for k, v in main.__dict__.items() if not k.startswith("__")}
self._globals = {
k: v
for k, v in xs.items()
if hasattr(v, "__module__")
and v.__module__ in [f"models_{suffix}", "typing"]
}
|
set_security_params
Source code in fastagency/api/openapi/openapi.py
| def set_security_params(
self, security_params: BaseSecurityParameters, name: Optional[str] = None
) -> None:
if name is not None:
security = self._security.get(name)
if security is None:
raise ValueError(f"Security is not set for '{name}'")
for match_security in security:
if match_security.accept(security_params):
break
else:
raise ValueError(
f"Security parameters {security_params} do not match security {security}"
)
self._security_params[name] = security_params
|