Async API

Vodoo provides a full async API via AsyncOdooClient. It exposes the same domain namespaces as the sync client, but all methods are async and use httpx for non-blocking HTTP.

Quick Start

import asyncio
from vodoo import AsyncOdooClient, OdooConfig

config = OdooConfig(
    url="https://my.odoo.com",
    database="mydb",
    username="bot@example.com",
    password="api-key",
)

async def main():
    async with AsyncOdooClient(config) as client:
        partners = await client.search_read(
            "res.partner",
            domain=[["is_company", "=", True]],
            fields=["name", "email"],
            limit=10,
        )
        for p in partners:
            print(p["name"])

asyncio.run(main())

Domain Namespaces

The async client has the same namespace properties as the sync client. All namespace methods must be awaited:

async with AsyncOdooClient(config) as client:
    tickets = await client.helpdesk.list(limit=5)
    leads = await client.crm.list(domain=[["type", "=", "opportunity"]])
    await client.helpdesk.note(ticket_id=42, message="Checked via async API")

Concurrent Requests

The async API shines when you need to make multiple independent calls:

import asyncio

async with AsyncOdooClient(config) as client:
    # Run all three queries concurrently
    tickets, leads, tasks = await asyncio.gather(
        client.helpdesk.list(limit=10),
        client.crm.list(limit=10),
        client.tasks.list(limit=10),
    )

Reference

AsyncOdooClient

AsyncOdooClient

AsyncOdooClient(config: OdooConfig, *, transport: AsyncOdooTransport | None = None, auto_detect: bool = True)

Async Odoo client for external API access.

Wraps an AsyncOdooTransport to provide a convenient async interface. Supports both legacy JSON-RPC (Odoo 14-18) and JSON-2 API (Odoo 19+).

Can be used as an async context manager::

async with AsyncOdooClient(config) as client:
    records = await client.search_read("res.partner", limit=5)

Initialize async Odoo client.

PARAMETER	DESCRIPTION
config	Odoo configuration TYPE: OdooConfig
transport	Explicit transport instance (skips auto-detection). When auto_detect is True and no transport is given, detection happens lazily on first use. TYPE: AsyncOdooTransport | None DEFAULT: None
auto_detect	If True, probe JSON-2 first then fall back to legacy on first use. If False, use legacy directly. TYPE: bool DEFAULT: True

Source code in src/vodoo/aio/client.py

def __init__(
    self,
    config: OdooConfig,
    *,
    transport: AsyncOdooTransport | None = None,
    auto_detect: bool = True,
) -> None:
    """Initialize async Odoo client.

    Args:
        config: Odoo configuration
        transport: Explicit transport instance (skips auto-detection).
                   When *auto_detect* is ``True`` and no transport is
                   given, detection happens lazily on first use.
        auto_detect: If True, probe JSON-2 first then fall back to
                     legacy on first use. If False, use legacy directly.
    """
    self.config = config
    self.url = config.url.rstrip("/")
    self.db = config.database
    self.username = config.username
    self.password = config.password
    self._retry = config.retry_config
    self._extra_headers = config.http_headers

    self._transport: AsyncOdooTransport | None = transport
    self._auto_detect = auto_detect
    self._init_lock = asyncio.Lock()

    # Domain namespaces
    self.helpdesk = _make_helpdesk(self)
    self.crm = _make_crm(self)
    self.tasks = _make_tasks(self)
    self.projects = _make_projects(self)
    self.account_moves = _make_account_moves(self)
    self.knowledge = _make_knowledge(self)
    self.timer = _make_timer(self)
    self.security = _make_security(self)
    self.generic = _make_generic(self)

transport property

transport: AsyncOdooTransport

The underlying transport (raises if not yet initialised).

is_json2 property

is_json2: bool

Whether the client is using the JSON-2 API (Odoo 19+).

uid property

uid: int

Get authenticated user ID (transport must be initialised).

close async

close() -> None

Close the underlying HTTP client.

Source code in src/vodoo/aio/client.py

async def close(self) -> None:
    """Close the underlying HTTP client."""
    if self._transport is not None:
        await self._transport.close()

get_uid async

get_uid() -> int

Get authenticated user ID, authenticating if needed.

Source code in src/vodoo/aio/client.py

async def get_uid(self) -> int:
    """Get authenticated user ID, authenticating if needed."""
    transport = await self._ensure_transport()
    return await transport.get_uid()

execute async

execute(model: str, method: str, *args: Any, **kwargs: Any) -> Any

Execute a method on an Odoo model.

Source code in src/vodoo/aio/client.py

async def execute(
    self,
    model: str,
    method: str,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Execute a method on an Odoo model."""
    transport = await self._ensure_transport()
    return await transport.execute_kw(model, method, list(args), kwargs or None)

execute_sudo async

execute_sudo(model: str, method: str, user_id: int, *args: Any, **kwargs: Any) -> Any

Execute a method as another user using sudo.

Source code in src/vodoo/aio/client.py

async def execute_sudo(
    self,
    model: str,
    method: str,
    user_id: int,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Execute a method as another user using sudo."""
    if "context" not in kwargs:
        kwargs["context"] = {}
    kwargs["context"]["sudo_user_id"] = user_id
    return await self.execute(model, method, *args, **kwargs)

search async

search(model: str, domain: list[Any] | None = None, limit: int | None = None, offset: int = 0, order: str | None = None) -> list[int]

Search for records.

Source code in src/vodoo/aio/client.py

async def search(
    self,
    model: str,
    domain: list[Any] | None = None,
    limit: int | None = None,
    offset: int = 0,
    order: str | None = None,
) -> list[int]:
    """Search for records."""
    transport = await self._ensure_transport()
    return await transport.search(model, domain, limit, offset, order)

read async

read(model: str, ids: list[int], fields: list[str] | None = None) -> list[dict[str, Any]]

Read records by IDs.

Source code in src/vodoo/aio/client.py

async def read(
    self,
    model: str,
    ids: list[int],
    fields: list[str] | None = None,
) -> list[dict[str, Any]]:
    """Read records by IDs."""
    transport = await self._ensure_transport()
    return _normalize_false(await transport.read(model, ids, fields))

search_read async

search_read(model: str, domain: list[Any] | None = None, fields: list[str] | None = None, limit: int | None = None, offset: int = 0, order: str | None = None) -> list[dict[str, Any]]

Search and read records in one call.

Source code in src/vodoo/aio/client.py

async def search_read(
    self,
    model: str,
    domain: list[Any] | None = None,
    fields: list[str] | None = None,
    limit: int | None = None,
    offset: int = 0,
    order: str | None = None,
) -> list[dict[str, Any]]:
    """Search and read records in one call."""
    transport = await self._ensure_transport()
    return _normalize_false(
        await transport.search_read(model, domain, fields, limit, offset, order)
    )

create async

create(model: str, values: dict[str, Any], context: dict[str, Any] | None = None) -> int

Create a new record.

Source code in src/vodoo/aio/client.py

async def create(
    self,
    model: str,
    values: dict[str, Any],
    context: dict[str, Any] | None = None,
) -> int:
    """Create a new record."""
    transport = await self._ensure_transport()
    return await transport.create(model, process_values(values), context)

write async

write(model: str, ids: list[int], values: dict[str, Any]) -> bool

Update records.

Source code in src/vodoo/aio/client.py

async def write(
    self,
    model: str,
    ids: list[int],
    values: dict[str, Any],
) -> bool:
    """Update records."""
    transport = await self._ensure_transport()
    return await transport.write(model, ids, process_values(values))

unlink async

unlink(model: str, ids: list[int]) -> bool

Delete records.

Source code in src/vodoo/aio/client.py

async def unlink(
    self,
    model: str,
    ids: list[int],
) -> bool:
    """Delete records."""
    transport = await self._ensure_transport()
    return await transport.unlink(model, ids)

fields_get async

fields_get(model: str, fields: list[str] | None = None, attributes: list[str] | None = None) -> dict[str, Any]

Return field definitions for a model.

PARAMETER	DESCRIPTION
model	Odoo model name (e.g., 'res.partner') TYPE: str
fields	Optional list of field names to inspect. None returns all fields. TYPE: list[str] | None DEFAULT: None
attributes	Optional list of field attributes to return (e.g., ['string', 'type', 'required']). None returns all attributes. TYPE: list[str] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any]	Dictionary mapping field names to their attribute dicts.

Source code in src/vodoo/aio/client.py

async def fields_get(
    self,
    model: str,
    fields: list[str] | None = None,
    attributes: list[str] | None = None,
) -> dict[str, Any]:
    """Return field definitions for a model.

    Args:
        model: Odoo model name (e.g., ``'res.partner'``)
        fields: Optional list of field names to inspect.
                ``None`` returns all fields.
        attributes: Optional list of field attributes to return
                    (e.g., ``['string', 'type', 'required']``).
                    ``None`` returns all attributes.

    Returns:
        Dictionary mapping field names to their attribute dicts.
    """
    transport = await self._ensure_transport()
    args: list[Any] = [fields or []]
    kwargs: dict[str, Any] = {}
    if attributes is not None:
        kwargs["attributes"] = attributes
    result: dict[str, Any] = await transport.execute_kw(
        model, "fields_get", args, kwargs or None
    )
    return result

name_search async

name_search(model: str, name: str, domain: list[Any] | None = None, limit: int = 7) -> list[tuple[int, str]]

Autocomplete search returning (id, display_name) pairs.

Source code in src/vodoo/aio/client.py

async def name_search(
    self,
    model: str,
    name: str,
    domain: list[Any] | None = None,
    limit: int = 7,
) -> list[tuple[int, str]]:
    """Autocomplete search returning (id, display_name) pairs."""
    transport = await self._ensure_transport()
    return await transport.name_search(model, name, domain, limit)

AsyncOdooTransport

transport

Async Odoo JSON-RPC transport abstraction. - AsyncLegacyTransport: Odoo 14-18 using POST /jsonrpc with service/method/args envelope - AsyncJSON2Transport: Odoo 19+ using POST /json/2// with bearer token auth The convenience methods on the base class and the concrete transport implementations intentionally duplicate their sync counterparts. This is a deliberate trade-off: every alternative (unasync, code generation, runtime indirection) adds build or type-system complexity that outweighs the cost of ~260 lines of mechanical duplication in a library of this size.

AsyncOdooTransport

AsyncOdooTransport(url: str, database: str, username: str, password: str, *, timeout: int = 30, retry: RetryConfig | None = None, extra_headers: dict[str, str] | None = None)

Bases: ABC

Abstract base for async Odoo RPC transports.

Mirrors :class:vodoo.transport.OdooTransport with async methods.

Source code in src/vodoo/aio/transport.py

def __init__(
    self,
    url: str,
    database: str,
    username: str,
    password: str,
    *,
    timeout: int = 30,
    retry: RetryConfig | None = None,
    extra_headers: dict[str, str] | None = None,
) -> None:
    self.url = url.rstrip("/")
    self.database = database.strip()
    self.username = username.strip()
    self.password = password.strip()
    self.timeout = timeout
    self.retry = retry or DEFAULT_RETRY
    self._uid: int | None = None
    self._extra_headers = extra_headers or {}
    self._http = httpx.AsyncClient(timeout=timeout, headers=self._extra_headers)

get_uid async

get_uid() -> int

Get authenticated user ID, authenticating if needed.

Source code in src/vodoo/aio/transport.py

async def get_uid(self) -> int:
    """Get authenticated user ID, authenticating if needed."""
    if self._uid is None:
        self._uid = await self.authenticate()
    return self._uid

authenticate abstractmethod async

authenticate() -> int

Authenticate and return the user ID.

RAISES	DESCRIPTION
AuthenticationError	If authentication fails.

Source code in src/vodoo/aio/transport.py

@abstractmethod
async def authenticate(self) -> int:
    """Authenticate and return the user ID.

    Raises:
        AuthenticationError: If authentication fails.
    """

execute_kw abstractmethod async

execute_kw(model: str, method: str, args: list[Any], kwargs: dict[str, Any] | None = None) -> Any

Execute a method on an Odoo model (execute_kw equivalent).

Source code in src/vodoo/aio/transport.py

@abstractmethod
async def execute_kw(
    self,
    model: str,
    method: str,
    args: list[Any],
    kwargs: dict[str, Any] | None = None,
) -> Any:
    """Execute a method on an Odoo model (execute_kw equivalent)."""

call_service abstractmethod async

call_service(service: str, method: str, args: list[Any]) -> Any

Call a JSON-RPC service method (e.g. common/authenticate).

Source code in src/vodoo/aio/transport.py

@abstractmethod
async def call_service(
    self,
    service: str,
    method: str,
    args: list[Any],
) -> Any:
    """Call a JSON-RPC service method (e.g. common/authenticate)."""

close async

close() -> None

Close the underlying HTTP client.

Source code in src/vodoo/aio/transport.py

async def close(self) -> None:
    """Close the underlying HTTP client."""
    await self._http.aclose()

search_read async

search_read(model: str, domain: list[Any] | None = None, fields: list[str] | None = None, limit: int | None = None, offset: int = 0, order: str | None = None) -> list[dict[str, Any]]

Search and read records.

Source code in src/vodoo/aio/transport.py

async def search_read(
    self,
    model: str,
    domain: list[Any] | None = None,
    fields: list[str] | None = None,
    limit: int | None = None,
    offset: int = 0,
    order: str | None = None,
) -> list[dict[str, Any]]:
    """Search and read records."""
    kw: dict[str, Any] = {}
    if fields is not None:
        kw["fields"] = fields
    if limit is not None:
        kw["limit"] = limit
    if offset > 0:
        kw["offset"] = offset
    if order is not None:
        kw["order"] = order
    result: list[dict[str, Any]] = await self.execute_kw(
        model, "search_read", [domain or []], kw
    )
    return result

search async

search(model: str, domain: list[Any] | None = None, limit: int | None = None, offset: int = 0, order: str | None = None) -> list[int]

Search for record IDs.

Source code in src/vodoo/aio/transport.py

async def search(
    self,
    model: str,
    domain: list[Any] | None = None,
    limit: int | None = None,
    offset: int = 0,
    order: str | None = None,
) -> list[int]:
    """Search for record IDs."""
    kw: dict[str, Any] = {}
    if limit is not None:
        kw["limit"] = limit
    if offset > 0:
        kw["offset"] = offset
    if order is not None:
        kw["order"] = order
    result: list[int] = await self.execute_kw(model, "search", [domain or []], kw)
    return result

read async

read(model: str, ids: list[int], fields: list[str] | None = None) -> list[dict[str, Any]]

Read records by IDs.

Source code in src/vodoo/aio/transport.py

async def read(
    self,
    model: str,
    ids: list[int],
    fields: list[str] | None = None,
) -> list[dict[str, Any]]:
    """Read records by IDs."""
    if fields is not None:
        result: list[dict[str, Any]] = await self.execute_kw(model, "read", [ids, fields])
    else:
        result = await self.execute_kw(model, "read", [ids])
    return result

create async

create(model: str, values: dict[str, Any], context: dict[str, Any] | None = None) -> int

Create a record and return its ID.

Source code in src/vodoo/aio/transport.py

async def create(
    self,
    model: str,
    values: dict[str, Any],
    context: dict[str, Any] | None = None,
) -> int:
    """Create a record and return its ID."""
    kw: dict[str, Any] = {}
    if context:
        kw["context"] = context
    result = await self.execute_kw(model, "create", [values], kw if kw else None)
    if isinstance(result, list) and len(result) == 1:
        return int(result[0])
    return int(result)

write async

write(model: str, ids: list[int], values: dict[str, Any]) -> bool

Update records.

Source code in src/vodoo/aio/transport.py

async def write(
    self,
    model: str,
    ids: list[int],
    values: dict[str, Any],
) -> bool:
    """Update records."""
    result: bool = await self.execute_kw(model, "write", [ids, values])
    return result

unlink async

unlink(model: str, ids: list[int]) -> bool

Delete records.

Source code in src/vodoo/aio/transport.py

async def unlink(
    self,
    model: str,
    ids: list[int],
) -> bool:
    """Delete records."""
    result: bool = await self.execute_kw(model, "unlink", [ids])
    return result

name_search async

name_search(model: str, name: str, domain: list[Any] | None = None, limit: int = 7) -> list[tuple[int, str]]

Autocomplete search returning (id, display_name) pairs.

Source code in src/vodoo/aio/transport.py

async def name_search(
    self,
    model: str,
    name: str,
    domain: list[Any] | None = None,
    limit: int = 7,
) -> list[tuple[int, str]]:
    """Autocomplete search returning (id, display_name) pairs."""
    result = await self.execute_kw(
        model,
        "name_search",
        [],
        {"name": name, "args": domain or [], "limit": limit},
    )
    return _parse_name_search(result)

AsyncLegacyTransport

AsyncLegacyTransport(url: str, database: str, username: str, password: str, *, timeout: int = 30, retry: RetryConfig | None = None, extra_headers: dict[str, str] | None = None)

Bases: AsyncOdooTransport

Async Odoo 14-18 legacy JSON-RPC transport.

Source code in src/vodoo/aio/transport.py

def __init__(
    self,
    url: str,
    database: str,
    username: str,
    password: str,
    *,
    timeout: int = 30,
    retry: RetryConfig | None = None,
    extra_headers: dict[str, str] | None = None,
) -> None:
    self.url = url.rstrip("/")
    self.database = database.strip()
    self.username = username.strip()
    self.password = password.strip()
    self.timeout = timeout
    self.retry = retry or DEFAULT_RETRY
    self._uid: int | None = None
    self._extra_headers = extra_headers or {}
    self._http = httpx.AsyncClient(timeout=timeout, headers=self._extra_headers)

AsyncJSON2Transport

AsyncJSON2Transport(url: str, database: str, username: str, password: str, *, timeout: int = 30, retry: RetryConfig | None = None, extra_headers: dict[str, str] | None = None)

Bases: AsyncOdooTransport

Async Odoo 19+ JSON-2 API transport.

Source code in src/vodoo/aio/transport.py

def __init__(
    self,
    url: str,
    database: str,
    username: str,
    password: str,
    *,
    timeout: int = 30,
    retry: RetryConfig | None = None,
    extra_headers: dict[str, str] | None = None,
) -> None:
    self.url = url.rstrip("/")
    self.database = database.strip()
    self.username = username.strip()
    self.password = password.strip()
    self.timeout = timeout
    self.retry = retry or DEFAULT_RETRY
    self._uid: int | None = None
    self._extra_headers = extra_headers or {}
    self._http = httpx.AsyncClient(timeout=timeout, headers=self._extra_headers)