Proxy class to generate client from OpenAPI schema.
Source code in fastagency/api/openapi/client.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, BaseSecurity] = {}
self.security_params: dict[Optional[str], BaseSecurityParameters] = {}
|
globals instance-attribute
kwargs instance-attribute
registered_funcs instance-attribute
security instance-attribute
security_params instance-attribute
servers instance-attribute
create classmethod
Source code in fastagency/api/openapi/client.py
| @classmethod
def create(
cls, openapi_json: Optional[str] = None, openapi_url: Optional[str] = None
) -> "OpenAPI":
if openapi_json is None and 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
with tempfile.TemporaryDirectory() as temp_dir:
td = Path(temp_dir)
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/client.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/client.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,
)
# 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 open(main_path) as f: # noqa: PTH123
main_py_code = f.read()
main_py_code = main_py_code.replace(
"from .models import", f"from models_{output_dir.name} import"
)
with open(main_path, "w") as f: # noqa: PTH123
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/client.py
| def get(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("get", path, **kwargs)
|
post
Source code in fastagency/api/openapi/client.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/client.py
| def put(self, path: str, **kwargs: Any) -> Callable[..., dict[str, Any]]:
return self._request("put", path, **kwargs)
|
register_for_execution
register_for_execution(agent: ConversableAgent) -> None
Source code in fastagency/api/openapi/client.py
| def register_for_execution(self, agent: "ConversableAgent") -> None:
for f in self.registered_funcs:
agent.register_for_execution()(f)
|
register_for_llm
register_for_llm(agent: ConversableAgent) -> None
Source code in fastagency/api/openapi/client.py
| def register_for_llm(self, agent: "ConversableAgent") -> None:
with add_to_globals(self.globals):
for f in self.registered_funcs:
agent.register_for_llm()(f)
|
set_globals
Source code in fastagency/api/openapi/client.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/client.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}'")
if not security.accept(security_params):
raise ValueError(
f"Security parameters {security_params} do not match security {security}"
)
self.security_params[name] = security_params
|