Skip to content

OpenAPI

fastagency.api.openapi.client.OpenAPI #

OpenAPI(
    servers: list[dict[str, Any]],
    title: Optional[str] = None,
    **kwargs: Any
)

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, list[BaseSecurity]] = {}
    self.security_params: dict[Optional[str], BaseSecurityParameters] = {}

globals instance-attribute #

globals: dict[str, Any] = {}

kwargs instance-attribute #

kwargs = kwargs

registered_funcs instance-attribute #

registered_funcs: list[Callable[..., Any]] = []

security instance-attribute #

security: dict[str, list[BaseSecurity]] = {}

security_params instance-attribute #

security_params: dict[
    Optional[str], BaseSecurityParameters
] = {}

servers instance-attribute #

servers = servers

title instance-attribute #

title = title

create classmethod #

create(
    openapi_json: Optional[str] = None,
    openapi_url: Optional[str] = None,
) -> OpenAPI
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 #

delete(
    path: str, **kwargs: Any
) -> Callable[..., dict[str, Any]]
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 #

generate_code(
    input_text: str,
    output_dir: Path,
    disable_timestamp: bool = False,
    custom_visitors: Optional[list[Path]] = None,
) -> str
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 #

get(
    path: str, **kwargs: Any
) -> Callable[..., dict[str, Any]]
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)

get_functions #

get_functions() -> list[str]
Source code in fastagency/api/openapi/client.py
def get_functions(self) -> list[str]:
    return [f.__name__ for f in self.registered_funcs]

post #

post(
    path: str, **kwargs: Any
) -> Callable[..., dict[str, Any]]
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 #

put(
    path: str, **kwargs: Any
) -> Callable[..., dict[str, Any]]
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)

set_globals #

set_globals(main: ModuleType, suffix: str) -> None
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 #

set_security_params(
    security_params: BaseSecurityParameters,
    name: Optional[str] = None,
) -> None
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}'")

        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