Skip to Content
Python RPC 🧪

Python RPC Client (Experimental)

Generate the Python client with vovk generate using the py or pySrc template.

Create a Python package with the CLI:

npx vovk generate --from py --out ./python_package

This produces:

      • __init__.py
      • api_client.py
      • py.typed
      • schema.json
    • pyproject.toml
    • setup.cfg
    • README.md

Publish to PyPI  with:

python3 -m build ./python_package --wheel --sdist && python3 -m twine upload ./python_package/dist/*

If you prefer generating source files to embed in another Python project, use the pySrc template:

npx vovk generate --from pySrc --out ./python_src

This generates:

    • __init__.py
    • api_client.py
    • py.typed
    • schema.json

Configuring the Python Client

You can configure generation so the client is produced automatically by the default generate command (no flags) and during vovk dev, which performs “hot generation” on schema changes. Add the py template to the composed client config:

vovk.config.mjs
/** @type {import('vovk').VovkConfig} */
const config = {
  composedClient: {
    fromTemplates: ['cjs', 'mjs', 'py'], // keeps the default "cjs" and "mjs" templates
  },
};
export default config;

The py template (and others) has a default outDir (./dist_python) for composed clients. Override it via template definitions:

vovk.config.mjs
/** @type {import('vovk').VovkConfig} */
const config = {
  // ...
  clientTemplateDefs: {
    py: {
      extends: 'py', // extends the built-in "py" template
      composedClient: {
        outDir: './my_dist_python', // custom output directory for the composed client
      },
    },
  },
};
export default config;

Generated Python Client Example

JSON Endpoints

The snippets below are adapted from a real example described on the Hello World page.

A controller like this:

UserController.ts
UserService.ts
1import { z } from "zod";
2import { prefix, post, operation } from "vovk";
3import { withZod } from "vovk-zod";
4import UserService from "./UserService";
5
6@prefix("users")
7export default class UserController {
8  @operation({
9    summary: "Update user",
10    description: "Update user by ID",
11  })
12  @post("{id}")
13  static updateUser = withZod({
14    body: z
15      .object({
16        email: z.email().meta({
17          description: "User email",
18          examples: ["john@example.com", "jane@example.com"],
19        }),
20        profile: z
21          .object({
22            name: z.string().meta({
23              description: "User full name",
24              examples: ["John Doe", "Jane Smith"],
25            }),
26            age: z
27              .int()
28              .min(16)
29              .max(120)
30              .meta({ description: "User age", examples: [25, 30] }),
31          })
32          .meta({ description: "User profile object" }),
33      })
34      .meta({ description: "User data object" }),
35    params: z
36      .object({
37        id: z.uuid().meta({
38          description: "User ID",
39          examples: ["123e4567-e89b-12d3-a456-426614174000"],
40        }),
41      })
42      .meta({
43        description: "Path parameters",
44      }),
45    query: z
46      .object({
47        notify: z
48          .enum(["email", "push", "none"])
49          .meta({ description: "Notification type" }),
50      })
51      .meta({
52        description: "Query parameters",
53      }),
54    output: z
55      .object({
56        success: z.boolean().meta({ description: "Success status" }),
57      })
58      .meta({ description: "Response object" }),
59    async handle(req, { id }) {
60      const body = await req.json();
61      const notify = req.nextUrl.searchParams.get("notify");
62
63      return UserService.updateUser(id, body, notify);
64    },
65  });
66}
1import { z } from "zod";
2import { prefix, post, operation } from "vovk";
3import { withZod } from "vovk-zod";
4import UserService from "./UserService";
5
6@prefix("users")
7export default class UserController {
8  @operation({
9    summary: "Update user",
10    description: "Update user by ID",
11  })
12  @post("{id}")
13  static updateUser = withZod({
14    body: z
15      .object({
16        email: z.email().meta({
17          description: "User email",
18          examples: ["john@example.com", "jane@example.com"],
19        }),
20        profile: z
21          .object({
22            name: z.string().meta({
23              description: "User full name",
24              examples: ["John Doe", "Jane Smith"],
25            }),
26            age: z
27              .int()
28              .min(16)
29              .max(120)
30              .meta({ description: "User age", examples: [25, 30] }),
31          })
32          .meta({ description: "User profile object" }),
33      })
34      .meta({ description: "User data object" }),
35    params: z
36      .object({
37        id: z.uuid().meta({
38          description: "User ID",
39          examples: ["123e4567-e89b-12d3-a456-426614174000"],
40        }),
41      })
42      .meta({
43        description: "Path parameters",
44      }),
45    query: z
46      .object({
47        notify: z
48          .enum(["email", "push", "none"])
49          .meta({ description: "Notification type" }),
50      })
51      .meta({
52        description: "Query parameters",
53      }),
54    output: z
55      .object({
56        success: z.boolean().meta({ description: "Success status" }),
57      })
58      .meta({ description: "Response object" }),
59    async handle(req, { id }) {
60      const body = await req.json();
61      const notify = req.nextUrl.searchParams.get("notify");
62
63      return UserService.updateUser(id, body, notify);
64    },
65  });
66}

…emits a Vovk.ts schema, which is then used to generate the Python client, following Python conventions, adding comments from description, and selecting appropriate number types. For example, age is generated as int, matching the controller definition.

./dist_python/src/package_name/__init__.py
from __future__ import annotations
from typing import Any, Dict, List, Literal, Optional, Set, TypedDict, Union, Tuple, Generator # type: ignore
from .api_client import ApiClient, HttpException
 
HttpException = HttpException
 
client = ApiClient('https://hello-world.vovk.dev/api')
 
class UserRPC: 
    # UserRPC.update_user POST `https://hello-world.vovk.dev/api/users/{id}`
    class __UpdateUserBody_profile(TypedDict):
        """
        User profile object
        """
        name: str
        age: int
    class UpdateUserBody(TypedDict):
        """
        User data object
        """
        email: str
        profile: UserRPC.__UpdateUserBody_profile
    class UpdateUserQuery(TypedDict):
        """
        Query parameters
        """
        notify: Literal["email", "push", "none"]
    class UpdateUserParams(TypedDict):
        """
        Path parameters
        """
        id: str
    class UpdateUserOutput(TypedDict):
        """
        Response object
        """
        success: bool
    @staticmethod
    def update_user(
        body: UpdateUserBody,
        query: UpdateUserQuery,
        params: UpdateUserParams,
        headers: Optional[Dict[str, str]] = None,
        files: Optional[Dict[str, Any]] = None,
        api_root: Optional[str] = None,
        disable_client_validation: bool = False
    ) -> UpdateUserOutput:
        """ 
        Update user
        Description: Update user by ID
        Body: User data object
        Query: Query parameters
        Returns: Response object
        """
        return client.request( # type: ignore
            segment_name='',
            rpc_name='UserRPC',
            handler_name='updateUser',
            body=body,
            query=query,
            params=params,
            headers=headers,
            files=files,
            api_root=api_root,
            disable_client_validation=disable_client_validation
        )

All RPC modules are generated in __init__.py, which contains the RPC functions and the associated types. Types for nested structures are emitted as TypedDicts.

Types for body, query, and params follow the pattern [PascalCaseMethodName][InputType]. For a method updateUser, you get UpdateUserBody, UpdateUserQuery, and UpdateUserParams.

from dist_python.src.vovk_hello_world import UserRPC
import vovk_hello_world
 
def main() -> None:
    body: UserRPC.UpdateUserBody = {
        "email": "john@example.com",
        "profile": {
            "name": "John Doe",
            "age": 25
        }
    }
    query: UserRPC.UpdateUserQuery = {"notify": "email"}
    params: UserRPC.UpdateUserParams = {"id": "123e4567-e89b-12d3-a456-426614174000"}
    # Update user using local module
    update_user_response = UserRPC.update_user(
        params=params,
        body=body,
        query=query
    )
    print('UserRPC.update_user:', update_user_response)
  
if __name__ == "__main__":
  try:
      main()
  except Exception as e:
      print(f"Error: {e}")

Under the hood, it uses requests  for HTTP, jsonschema  for client-side validation, and other common libraries.

JSON Lines Endpoints

For continuous streaming with JSON Lines endpoints, the Python client returns a Generator you can iterate over.

A controller like this:

StreamController.ts
StreamService.ts
1import { prefix, get, operation } from "vovk";
2import { withZod } from "vovk-zod";
3import { z } from "zod";
4import StreamService from "./StreamService";
5
6@prefix("streams")
7export default class StreamController {
8  @operation({
9    summary: "Stream tokens",
10    description: "Stream tokens to the client",
11  })
12  @get("tokens")
13  static streamTokens = withZod({
14    iteration: z
15      .object({
16        message: z.string().meta({ description: "Message from the token" }),
17      })
18      .meta({
19        description: "Streamed token object",
20      }),
21    async *handle() {
22      yield* StreamService.streamTokens();
23    },
24  });
25}
1import { prefix, get, operation } from "vovk";
2import { withZod } from "vovk-zod";
3import { z } from "zod";
4import StreamService from "./StreamService";
5
6@prefix("streams")
7export default class StreamController {
8  @operation({
9    summary: "Stream tokens",
10    description: "Stream tokens to the client",
11  })
12  @get("tokens")
13  static streamTokens = withZod({
14    iteration: z
15      .object({
16        message: z.string().meta({ description: "Message from the token" }),
17      })
18      .meta({
19        description: "Streamed token object",
20      }),
21    async *handle() {
22      yield* StreamService.streamTokens();
23    },
24  });
25}

Compiles to:

./dist_python/src/package_name/__init__.py
class StreamRPC: 
    # StreamRPC.stream_tokens GET `https://hello-world.vovk.dev/api/streams/tokens`
    class StreamTokensIteration(TypedDict):
        """
        Streamed token object
        """
        message: str
    @staticmethod
    def stream_tokens(
 
        headers: Optional[Dict[str, str]] = None,
        files: Optional[Dict[str, Any]] = None,
        api_root: Optional[str] = None,
        disable_client_validation: bool = False
    ) -> Generator[StreamTokensIteration, None, None]:
        """ 
        Stream tokens
        Description: Stream tokens to the client
        """
        return client.request( # type: ignore
            segment_name='',
            rpc_name='StreamRPC',
            handler_name='streamTokens',
 
            headers=headers,
            files=files,
            api_root=api_root,
            disable_client_validation=disable_client_validation
        )

Usage:

from dist_python.src.vovk_hello_world import StreamRPC  # local module
import vovk_hello_world
 
def main() -> None:
    stream_response = StreamRPC.stream_tokens()
    print("streamTokens:")
    for item in stream_response:
        print(item['message'], end='', flush=True)
 
if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"Error: {e}")

Roadmap

  • 🐞 Allow circular $refs with OpenAPI Mixins.
  • ✨ Generate importable types for named schemas defined in components/schemas.
Last updated on
Imports CustomizationRust RPC 🧪