chore: add XAPI SMAPIv3 API stubs

Signed-off-by: Ronan Abhamon <ronan.abhamon@vates.tech>

Author: Wescoeur

pyproject.toml

@@ -60,6 +60,47 @@ disable_error_code = [
 ]
 disallow_untyped_defs = false
 
+# To generate XAPI SMAPIv3 API stubs, run the following commands in an XCP-ng environment:
+#
+# # Clone xen-api.
+# yum install git
+# git clone https://github.com/xapi-project/xen-api
+# cd xen-api/
+#
+# # Init build env.
+# yum install opam gcc make
+# export OCAML_VERSION_FULL="4.14.2"
+# opam init
+# opam switch create xen-api ocaml-base-compiler.$OCAML_VERSION_FULL
+# eval $(opam env --switch=xen-api --set-switch)
+# opam repo add xs-opam https://github.com/xapi-project/xs-opam.git
+# opam repo remove default
+#
+# # Install deps.
+# opam install dune rpclib ppx_deriving_rpc rresult lwt fmt alcotest -y
+#
+# # Build.
+# dune build ocaml/xapi-storage
+#
+# # Build python API.
+# cd _build/default/ocaml/xapi-storage/python/
+# make
+#
+# # Clean env.
+# opam clean
+# rm -rf /root/.opam
+#
+# # Copy python dir in your dev env and then:
+# stubgen --include-private python/ -o <XCP_STORAGE_PATH>/stubs/
+# # Replace invalid return types caused by "raise Unimplemented" calls.
+# find <XCP_STORAGE_PATH>/stubs/xapi/ -name "*.pyi" -exec sed -i 's/ -> None:/:/g' {} +
+[[tool.mypy.overrides]]
+module = "xapi.*"
+disable_error_code = [
+    "explicit-override",
+    "no-untyped-def"
+]
+
 [tool.pyrefly]
 python-version = "3.14"
 search-path = [".", "stubs"]
@@ -138,7 +179,7 @@ case-sensitive = false
 combine-as-imports = true
 force-sort-within-sections = true
 known-first-party = ["xcp_storage"]
-known-third-party = ["linstor"]
+known-third-party = ["linstor", "xapi"]
 lines-after-imports = 1
 order-by-type = false
 section-order = [

stubs/xapi/__init__.pyi

@@ -0,0 +1,41 @@
+import argparse
+from _typeshed import Incomplete
+
+def success(result): ...
+def handle_exception(e, code=None, params=None): ...
+
+class XenAPIException(Exception):
+    code: Incomplete
+    params: Incomplete
+    def __init__(self, code, params): ...
+
+class MissingDependency(Exception):
+    missing: Incomplete
+    def __init__(self, missing): ...
+    def __str__(self) -> str: ...
+
+class Rpc_light_failure(Exception):
+    name: Incomplete
+    args: Incomplete
+    def __init__(self, name, args): ...
+    def failure(self): ...
+
+class Unimplemented(Rpc_light_failure):
+    def __init__(self, name): ...
+
+class InternalError(Rpc_light_failure):
+    def __init__(self, error): ...
+
+class UnmarshalException(InternalError):
+    def __init__(self, thing, ty, desc): ...
+
+class TypeError(InternalError):
+    def __init__(self, expected, actual): ...
+
+class UnknownMethod(InternalError):
+    def __init__(self, name): ...
+
+def is_long(x): ...
+
+class ListAction(argparse.Action):
+    def __call__(self, parser, namespace, values, option_string=None): ...

stubs/xapi/storage/__init__.pyi

@@ -0,0 +1,121 @@
+from xapi import *
+from _typeshed import Incomplete
+
+get_str = str
+long = int
+unicode = str
+str = bytes
+
+class Datapath_server_dispatcher:
+    _impl: Incomplete
+    _dispatcher_dict: Incomplete
+    def __init__(self, impl): ...
+    def open(self, args): ...
+    def attach(self, args): ...
+    def activate(self, args): ...
+    def activate_readonly(self, args): ...
+    def import_activate(self, args): ...
+    def deactivate(self, args): ...
+    def detach(self, args): ...
+    def close(self, args): ...
+    def _dispatch(self, method, params): ...
+
+class Datapath_skeleton:
+    def __init__(self): ...
+    def open(self, dbg, uri, persistent): ...
+    def attach(self, dbg, uri, domain): ...
+    def activate(self, dbg, uri, domain): ...
+    def activate_readonly(self, dbg, uri, domain): ...
+    def import_activate(self, dbg, uri, domain): ...
+    def deactivate(self, dbg, uri, domain): ...
+    def detach(self, dbg, uri, domain): ...
+    def close(self, dbg, uri): ...
+
+class Datapath_test:
+    def __init__(self): ...
+    def open(self, dbg, uri, persistent): ...
+    def attach(self, dbg, uri, domain): ...
+    def activate(self, dbg, uri, domain): ...
+    def activate_readonly(self, dbg, uri, domain): ...
+    def import_activate(self, dbg, uri, domain): ...
+    def deactivate(self, dbg, uri, domain): ...
+    def detach(self, dbg, uri, domain): ...
+    def close(self, dbg, uri): ...
+
+class Datapath_commandline:
+    impl: Incomplete
+    dispatcher: Incomplete
+    def __init__(self, impl): ...
+    def _parse_open(self): ...
+    def _parse_attach(self): ...
+    def _parse_activate(self): ...
+    def _parse_activate_readonly(self): ...
+    def _parse_import_activate(self): ...
+    def _parse_deactivate(self): ...
+    def _parse_detach(self): ...
+    def _parse_close(self): ...
+    def open(self): ...
+    def attach(self): ...
+    def activate(self): ...
+    def activate_readonly(self): ...
+    def import_activate(self): ...
+    def deactivate(self): ...
+    def detach(self): ...
+    def close(self): ...
+
+class Data_server_dispatcher:
+    _impl: Incomplete
+    _dispatcher_dict: Incomplete
+    def __init__(self, impl): ...
+    def copy(self, args): ...
+    def mirror(self, args): ...
+    def stat(self, args): ...
+    def cancel(self, args): ...
+    def destroy(self, args): ...
+    def ls(self, args): ...
+    def _dispatch(self, method, params): ...
+
+class Data_skeleton:
+    def __init__(self): ...
+    def copy(self, dbg, uri, domain, remote, blocklist): ...
+    def mirror(self, dbg, uri, domain, remote): ...
+    def stat(self, dbg, operation): ...
+    def cancel(self, dbg, operation): ...
+    def destroy(self, dbg, operation): ...
+    def ls(self, dbg): ...
+
+class Data_test:
+    def __init__(self): ...
+    def copy(self, dbg, uri, domain, remote, blocklist): ...
+    def mirror(self, dbg, uri, domain, remote): ...
+    def stat(self, dbg, operation): ...
+    def cancel(self, dbg, operation): ...
+    def destroy(self, dbg, operation): ...
+    def ls(self, dbg): ...
+
+class Data_commandline:
+    impl: Incomplete
+    dispatcher: Incomplete
+    def __init__(self, impl): ...
+    def _parse_copy(self): ...
+    def _parse_mirror(self): ...
+    def _parse_stat(self): ...
+    def _parse_cancel(self): ...
+    def _parse_destroy(self): ...
+    def _parse_ls(self): ...
+    def copy(self): ...
+    def mirror(self): ...
+    def stat(self): ...
+    def cancel(self): ...
+    def destroy(self): ...
+    def ls(self): ...
+
+class datapath_server_dispatcher:
+    Datapath: Incomplete
+    Data: Incomplete
+    _dispatcher_dict: Incomplete
+    def __init__(self, Datapath=None, Data=None): ...
+    def _dispatch(self, method, params): ...
+
+class datapath_server_test(datapath_server_dispatcher):
+    def __init__(self): ...

stubs/xapi/storage/api/__init__.pyi

@@ -0,0 +1,48 @@
+from xapi import *
+from _typeshed import Incomplete
+
+get_str = str
+long = int
+unicode = str
+str = bytes
+
+class Plugin_server_dispatcher:
+    _impl: Incomplete
+    _dispatcher_dict: Incomplete
+    def __init__(self, impl): ...
+    def query(self, args): ...
+    def ls(self, args): ...
+    def diagnostics(self, args): ...
+    def _dispatch(self, method, params): ...
+
+class Plugin_skeleton:
+    def __init__(self): ...
+    def query(self, dbg): ...
+    def ls(self, dbg): ...
+    def diagnostics(self, dbg): ...
+
+class Plugin_test:
+    def __init__(self): ...
+    def query(self, dbg): ...
+    def ls(self, dbg): ...
+    def diagnostics(self, dbg): ...
+
+class Plugin_commandline:
+    impl: Incomplete
+    dispatcher: Incomplete
+    def __init__(self, impl): ...
+    def _parse_query(self): ...
+    def _parse_ls(self): ...
+    def _parse_diagnostics(self): ...
+    def query(self): ...
+    def ls(self): ...
+    def diagnostics(self): ...
+
+class plugin_server_dispatcher:
+    Plugin: Incomplete
+    _dispatcher_dict: Incomplete
+    def __init__(self, Plugin=None): ...
+    def _dispatch(self, method, params): ...
+
+class plugin_server_test(plugin_server_dispatcher):
+    def __init__(self): ...

stubs/xapi/storage/api/v5/__init__.pyi

@@ -0,0 +1,53 @@
+from xapi import *
+from _typeshed import Incomplete
+
+get_str = str
+long = int
+unicode = str
+str = bytes
+
+class Task_server_dispatcher:
+    _impl: Incomplete
+    _dispatcher_dict: Incomplete
+    def __init__(self, impl): ...
+    def stat(self, args): ...
+    def cancel(self, args): ...
+    def destroy(self, args): ...
+    def ls(self, args): ...
+    def _dispatch(self, method, params): ...
+
+class Task_skeleton:
+    def __init__(self): ...
+    def stat(self, dbg, id): ...
+    def cancel(self, dbg, id): ...
+    def destroy(self, dbg, id): ...
+    def ls(self, dbg): ...
+
+class Task_test:
+    def __init__(self): ...
+    def stat(self, dbg, id): ...
+    def cancel(self, dbg, id): ...
+    def destroy(self, dbg, id): ...
+    def ls(self, dbg): ...
+
+class Task_commandline:
+    impl: Incomplete
+    dispatcher: Incomplete
+    def __init__(self, impl): ...
+    def _parse_stat(self): ...
+    def _parse_cancel(self): ...
+    def _parse_destroy(self): ...
+    def _parse_ls(self): ...
+    def stat(self): ...
+    def cancel(self): ...
+    def destroy(self): ...
+    def ls(self): ...
+
+class task_server_dispatcher:
+    Task: Incomplete
+    _dispatcher_dict: Incomplete
+    def __init__(self, Task=None): ...
+    def _dispatch(self, method, params): ...
+
+class task_server_test(task_server_dispatcher):
+    def __init__(self): ...