Skip to content

OpenAPI

fastagency.api.openapi.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/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] = {}

function_names property #

function_names: list[str]

create classmethod #

create(
    *,
    openapi_json: Optional[str] = None,
    openapi_url: Optional[str] = None,
    client_source_path: Optional[str] = None,
    servers: Optional[list[dict[str, Any]]] = None
) -> OpenAPI
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 #

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

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/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 #

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

get_function(name: str) -> Callable[..., dict[str, Any]]
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 #

get_functions() -> list[str]
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 #

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

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

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

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

set_function(
    name: str, func: Callable[..., dict[str, Any]]
) -> None
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 #

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

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