Skip to Content
Python Client

Python Client

vovk generate creates the Python client from the py or pySrc template.

Install the generator package:

npm install vovk-python --save-dev

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

To generate source files for another Python project instead, 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 the default generate command (without flags) to create the client. vovk dev creates it too, and creates it again on every schema change. Add the py template to the composed client config:

vovk.config.mjs
/** @type {import('vovk').VovkConfig} */
const config = {
  composedClient: {
    fromTemplates: ['ts', 'py'], // keeps the default "ts" template
  },
  clientTemplateDefs: {
    py: {
      extends: 'py', // extends the built-in "py" template
      outputConfig: {
        origin: 'https://example.com', // the server the Python client calls
      },
    },
  },
};
export default config;

The Python client calls an absolute URL, so it needs outputConfig.origin. Set it on the py template, and the TypeScript client keeps its relative URLs. Without an origin, vovk generate warns, and every call needs its own api_root.

Like other templates, the py template has a default outDir for composed clients: ./dist_python. Change it in the 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 samples below are based on the example from the Hello World page.

A controller like this:

src/modules/user/user-controller.ts
import { operation, post, prefix, procedure } from 'vovk';
import { z } from 'zod';
import UserService from './user-service';
 
@prefix('users')
export default class UserController {
  @operation({
    summary: 'Update user',
    description: 'Update user by ID',
  })
  @post('{id}')
  static updateUser = procedure({
    body: z
      .object({
        email: z.email().meta({
          description: 'User email',
          examples: ['john@example.com', 'jane@example.com'],
        }),
        profile: z
          .object({
            name: z
              .string()
              .min(2)
              .meta({
                description: 'User full name',
                examples: ['John Doe', 'Jane Smith'],
              }),
            age: z
              .int()
              .min(16)
              .max(120)
              .meta({ description: 'User age', examples: [25, 30] }),
          })
          .meta({ description: 'User profile object' }),
      })
      .meta({ description: 'User data object' }),
    params: z
      .object({
        id: z.uuid().meta({
          description: 'User ID',
          examples: ['123e4567-e89b-12d3-a456-426614174000'],
        }),
      })
      .meta({
        description: 'Path parameters',
      }),
    query: z
      .object({
        notify: z
          .enum(['email', 'push', 'none'])
          .meta({ description: 'Notification type' }),
      })
      .meta({
        description: 'Query parameters',
      }),
    output: z
      .object({
        success: z.boolean().meta({ description: 'Success status' }),
        id: z.uuid().meta({ description: 'User ID' }),
        notify: z.enum(['email', 'push', 'none']).meta({
          description: 'Notification type',
        }),
      })
      .meta({ description: 'Response object' }),
  }).handle(async (req, { id }) => {
    const body = await req.vovk.body();
    const { notify } = req.vovk.query();
 
    return UserService.updateUser(id, body, notify);
  });
}

The code above is fetched from GitHub repository. 

…emits a Vovk.ts schema. The Python client is generated from it. It follows Python conventions, adds comments from description, and picks matching number types: age becomes an int, as in the controller.

./dist_python/src/package_name/__init__.py
from __future__ import annotations
import sys
from typing import Any, Dict, List, Literal, Optional, Sequence, Set, Union, Tuple, Generator, BinaryIO # pyright: ignore[reportUnusedImport]
# as HttpException: a name the package exports, also under mypy --strict
from .api_client import ApiClient, HttpException as HttpException
 
# pydantic reads a TypedDict only from typing_extensions before Python 3.12
if sys.version_info >= (3, 12):
    from typing import NotRequired, TypeAlias, TypedDict
else:
    from typing_extensions import NotRequired, TypeAlias, TypedDict
 
client = ApiClient("https://hello-world.vovk.dev/api", {
    "": ("https://hello-world.vovk.dev/api", ""),
    "static": ("https://hello-world.vovk.dev/api", "static"),
})
 
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
        id: str
        notify: Literal["email", "push", "none"]
    @staticmethod
    def update_user(
        body: UpdateUserBody,
        query: UpdateUserQuery,
        params: UpdateUserParams,
        headers: Optional[Dict[str, str]] = 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[no-any-return, unused-ignore]
            segment_name="",
            rpc_name="UserRPC",
            handler_name="updateUser",
            body=body,
            query=query,
            params=params,
            headers=headers,
            api_root=api_root,
            disable_client_validation=disable_client_validation
        )

All RPC modules are in __init__.py, with the RPC functions and their types. Nested structures get TypedDict types. A key the schema doesn’t require is NotRequired. A type that isn’t a TypedDict, such as a str output, is a TypeAlias. Before Python 3.12, NotRequired, TypeAlias and TypedDict come from typing_extensions, where pydantic reads a TypedDict from.

Types for body, query and params are named [PascalCaseMethodName][InputType]. For updateUser, they are UpdateUserBody, UpdateUserQuery and UpdateUserParams.

Method names are in snake_case: UserRPC.updateUser is UserRPC.update_user. Each RPC class names its methods in schema order. A name that is already taken gets the first free suffix, _2, _3, so getUserByID and getUserById give get_user_by_id and get_user_by_id_2.

from vovk_hello_world import UserRPC
 
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_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}")

The client uses requests  for HTTP, jsonschema  for client-side validation, and other common libraries. A pattern that Python’s re can’t compile, such as \p{L}, is left to the server. A form body is checked as the object it is made from, without its files, and a uri as the server’s z.url() reads it.

A call returns the parsed JSON, or None for an empty JSON body. A text/* response, or a response of any type with a charset, comes back as a str, decoded as UTF-8 unless the charset names another encoding. Any other response, such as a file, comes back as bytes.

JSON Lines Endpoints

For JSON Lines endpoints, the Python client returns a Generator that you iterate over.

A controller like this:

src/modules/stream/stream-controller.ts
import { get, operation, prefix, procedure } from 'vovk';
import { z } from 'zod';
import StreamService from './stream-service';
 
@prefix('streams')
export default class StreamController {
  @operation({
    summary: 'Stream tokens',
    description: 'Stream tokens to the client',
  })
  @get('tokens')
  static streamTokens = procedure({
    validateEachIteration: true,
    iteration: z
      .object({
        message: z.string().meta({ description: 'Message from the token' }),
      })
      .meta({
        description: 'Streamed token object',
      }),
  }).handle(async function* () {
    yield* StreamService.streamTokens();
  });
}

The code above is fetched from GitHub repository. 

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,
        api_root: Optional[str] = None,
        disable_client_validation: bool = False,
        query: Optional[Dict[str, Any]] = None
    ) -> Generator[StreamTokensIteration, None, None]:
        """
        Stream tokens
        Description: Stream tokens to the client
        """
        return client.request( # type: ignore[no-any-return, unused-ignore]
            segment_name="",
            rpc_name="StreamRPC",
            handler_name="streamTokens",
            query=query,
            headers=headers,
            api_root=api_root,
            disable_client_validation=disable_client_validation
        )

Usage:

from vovk_hello_world import StreamRPC
 
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}")

Timeout and Session

All calls share one requests.Session, so connections stay open between them. The client waits 10 seconds to connect and 300 seconds for each read. An upload fails when the server takes no data for 10 seconds. The timeout and the session are attributes of the module-level client:

from package_name import client
 
client.timeout = 30  # seconds, or a (connect, read) tuple; None waits forever
client.session.headers["Authorization"] = "Bearer ..."

The session doesn’t keep cookies that responses set, because one client serves every call and thread. A cookie you set by hand, such as client.session.cookies.set("session", "..."), goes with every call. To keep the cookies that responses set, call client.session.cookies.set_policy(http.cookiejar.DefaultCookiePolicy()).

Roadmap/bugs

  • ✨ Generate importable types for named schemas defined in components/schemas.
Last updated on
OpenAPI MixinsRust Client