Transport

Transport abstraction for Odoo JSON-RPC and JSON-2 protocols.

transport

Odoo JSON-RPC transport abstraction.

Provides two implementations: - LegacyTransport: Odoo 14-18 using POST /jsonrpc with service/method/args envelope - JSON2Transport: Odoo 19+ using POST /json/2// with bearer token auth

OdooTransport

OdooTransport(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 Odoo RPC transports.

Each transport knows how to authenticate, call model methods, and perform CRUD operations. The public API is identical regardless of the underlying protocol (legacy JSON-RPC or JSON-2 REST).

Source code in src/vodoo/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.Client(timeout=timeout, headers=self._extra_headers)

uid property

uid: int

Get authenticated user ID, authenticating if needed.

authenticate abstractmethod

authenticate() -> int

Authenticate and return the user ID.

RAISES	DESCRIPTION
AuthenticationError	If authentication fails.

Source code in src/vodoo/transport.py

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

    Raises:
        AuthenticationError: If authentication fails.
    """

execute_kw abstractmethod

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).

PARAMETER	DESCRIPTION
model	Odoo model name (e.g., 'project.task') TYPE: str
method	Method name (e.g., 'search_read') TYPE: str
args	Positional arguments TYPE: list[Any]
kwargs	Keyword arguments TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Any	Method result

RAISES	DESCRIPTION
TransportError	On RPC/HTTP errors.

Source code in src/vodoo/transport.py

@abstractmethod
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).

    Args:
        model: Odoo model name (e.g., 'project.task')
        method: Method name (e.g., 'search_read')
        args: Positional arguments
        kwargs: Keyword arguments

    Returns:
        Method result

    Raises:
        TransportError: On RPC/HTTP errors.
    """

call_service abstractmethod

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

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

PARAMETER	DESCRIPTION
service	Service name ('common', 'object', 'db') TYPE: str
method	Method name TYPE: str
args	Arguments TYPE: list[Any]

RETURNS	DESCRIPTION
Any	Result

RAISES	DESCRIPTION
TransportError	On RPC/HTTP errors.

Source code in src/vodoo/transport.py

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

    Args:
        service: Service name ('common', 'object', 'db')
        method: Method name
        args: Arguments

    Returns:
        Result

    Raises:
        TransportError: On RPC/HTTP errors.
    """

close

close() -> None

Close the underlying HTTP client.

Source code in src/vodoo/transport.py

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

search_read

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/transport.py

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]] = self.execute_kw(model, "search_read", [domain or []], kw)
    return result

search

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/transport.py

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] = self.execute_kw(model, "search", [domain or []], kw)
    return result

read

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

Read records by IDs.

Source code in src/vodoo/transport.py

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]] = self.execute_kw(model, "read", [ids, fields])
    else:
        result = self.execute_kw(model, "read", [ids])
    return result

create

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/transport.py

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 = self.execute_kw(model, "create", [values], kw if kw else None)
    # JSON-2 returns a list of IDs (vals_list), unwrap single-record creates
    if isinstance(result, list) and len(result) == 1:
        return int(result[0])
    return int(result)

write

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

Update records.

Source code in src/vodoo/transport.py

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

unlink

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

Delete records.

Source code in src/vodoo/transport.py

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

name_search

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/transport.py

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 = self.execute_kw(
        model,
        "name_search",
        [],
        {"name": name, "args": domain or [], "limit": limit},
    )
    return _parse_name_search(result)

LegacyTransport

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

Bases: OdooTransport

Odoo 14-18 legacy JSON-RPC transport.

Uses POST /jsonrpc with service/method/args envelope.

Source code in src/vodoo/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.Client(timeout=timeout, headers=self._extra_headers)

JSON2Transport

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

Bases: OdooTransport

Odoo 19+ JSON-2 API transport.

Uses POST /json/2// with bearer token auth.

Source code in src/vodoo/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.Client(timeout=timeout, headers=self._extra_headers)