Async Base Operations

Async shared CRUD, messaging, and attachment helpers used by async domain namespaces.

base

Async base operations for Odoo models.

Provides async versions of I/O functions from :mod:vodoo.base. Pure display/formatting functions are re-exported unchanged.

configure_output

configure_output(*, console: Any = None, simple: bool = False, json_mode: bool = False, toon_mode: bool = False) -> None

Deprecated compatibility shim for :func:vodoo.cli.output.configure_output.

Source code in src/vodoo/base.py

def configure_output(
    *,
    console: Any = None,
    simple: bool = False,
    json_mode: bool = False,
    toon_mode: bool = False,
) -> None:
    """Deprecated compatibility shim for :func:`vodoo.cli.output.configure_output`."""
    _cli_output().configure_output(
        console=console,
        simple=simple,
        json_mode=json_mode,
        toon_mode=toon_mode,
    )

display_attachments

display_attachments(attachments: list[dict[str, Any]]) -> None

Deprecated compatibility shim for CLI attachment rendering.

Source code in src/vodoo/base.py

def display_attachments(attachments: list[dict[str, Any]]) -> None:
    """Deprecated compatibility shim for CLI attachment rendering."""
    _cli_display().display_attachments(attachments)

display_messages

display_messages(messages: list[dict[str, Any]], show_html: bool = False) -> None

Deprecated compatibility shim for CLI message rendering.

Source code in src/vodoo/base.py

def display_messages(messages: list[dict[str, Any]], show_html: bool = False) -> None:
    """Deprecated compatibility shim for CLI message rendering."""
    _cli_display().display_messages(messages, show_html)

display_record_detail

display_record_detail(record: dict[str, Any], *, show_html: bool = False, record_type: str = 'Record') -> None

Deprecated compatibility shim for CLI record-detail rendering.

Source code in src/vodoo/base.py

def display_record_detail(
    record: dict[str, Any],
    *,
    show_html: bool = False,
    record_type: str = "Record",
) -> None:
    """Deprecated compatibility shim for CLI record-detail rendering."""
    _cli_display().display_record_detail(
        record,
        show_html=show_html,
        record_type=record_type,
    )

display_records

display_records(records: list[dict[str, Any]], title: str = 'Records') -> None

Deprecated compatibility shim for CLI record rendering.

Source code in src/vodoo/base.py

def display_records(records: list[dict[str, Any]], title: str = "Records") -> None:
    """Deprecated compatibility shim for CLI record rendering."""
    _cli_display().display_records(records, title)

display_tags

display_tags(tags: list[dict[str, Any]], title: str = 'Tags') -> None

Deprecated compatibility shim for CLI tag rendering.

Source code in src/vodoo/base.py

def display_tags(tags: list[dict[str, Any]], title: str = "Tags") -> None:
    """Deprecated compatibility shim for CLI tag rendering."""
    _cli_display().display_tags(tags, title)

get_record_url

get_record_url(client: OdooClient | Any, model: str, record_id: int) -> str

Get the web URL for a record.

Uses the canonical path URL for a selected JSON-2 transport and the legacy hash URL otherwise. Works with sync and async clients without making an additional request.

PARAMETER	DESCRIPTION
client	Odoo client (sync or async) TYPE: OdooClient | Any
model	Model name TYPE: str
record_id	Record ID TYPE: int

RETURNS	DESCRIPTION
str	URL to view the record in Odoo web interface

Examples:

>>> get_record_url(client, "helpdesk.ticket", 42)
'https://odoo.example.com/web#id=42&model=helpdesk.ticket&view_type=form'

Source code in src/vodoo/base.py

def get_record_url(client: OdooClient | Any, model: str, record_id: int) -> str:
    """Get the web URL for a record.

    Uses the canonical path URL for a selected JSON-2 transport and the
    legacy hash URL otherwise. Works with sync and async clients without
    making an additional request.

    Args:
        client: Odoo client (sync or async)
        model: Model name
        record_id: Record ID

    Returns:
        URL to view the record in Odoo web interface

    Examples:
        >>> get_record_url(client, "helpdesk.ticket", 42)
        'https://odoo.example.com/web#id=42&model=helpdesk.ticket&view_type=form'

    """
    return build_record_url(
        client.config.url,
        model,
        record_id,
        "json2" if getattr(client, "is_json2", False) else "jsonrpc",
    )

list_records async

list_records(client: AsyncOdooClient, model: str, domain: list[Any] | None = None, limit: int | None = 50, fields: list[str] | None = None, order: str = 'create_date desc') -> list[dict[str, Any]]

List records from a model.

PARAMETER	DESCRIPTION
client	Async Odoo client TYPE: AsyncOdooClient
model	Model name TYPE: str
domain	Search domain filters TYPE: list[Any] | None DEFAULT: None
limit	Maximum number of records TYPE: int | None DEFAULT: 50
fields	List of fields to fetch TYPE: list[str] | None DEFAULT: None
order	Sort order TYPE: str DEFAULT: 'create_date desc'

RETURNS	DESCRIPTION
list[dict[str, Any]]	List of record dictionaries

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

async def list_records(
    client: AsyncOdooClient,
    model: str,
    domain: list[Any] | None = None,
    limit: int | None = 50,
    fields: list[str] | None = None,
    order: str = "create_date desc",
) -> list[dict[str, Any]]:
    """List records from a model.

    Args:
        client: Async Odoo client
        model: Model name
        domain: Search domain filters
        limit: Maximum number of records
        fields: List of fields to fetch
        order: Sort order

    Returns:
        List of record dictionaries
    """
    return await client.search_read(
        model,
        domain=domain,
        fields=fields,
        limit=limit,
        order=order,
    )

get_record async

get_record(client: AsyncOdooClient, model: str, record_id: int, fields: list[str] | None = None) -> dict[str, Any]

Get detailed record information.

PARAMETER	DESCRIPTION
client	Async Odoo client TYPE: AsyncOdooClient
model	Model name TYPE: str
record_id	Record ID TYPE: int
fields	List of field names to read TYPE: list[str] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any]	Record dictionary

RAISES	DESCRIPTION
RecordNotFoundError	If record not found

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

async def get_record(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    fields: list[str] | None = None,
) -> dict[str, Any]:
    """Get detailed record information.

    Args:
        client: Async Odoo client
        model: Model name
        record_id: Record ID
        fields: List of field names to read

    Returns:
        Record dictionary

    Raises:
        RecordNotFoundError: If record not found
    """
    records = await client.read(model, [record_id], fields=fields)
    if not records:
        raise RecordNotFoundError(model, record_id)
    return records[0]

list_fields async

list_fields(client: AsyncOdooClient, model: str) -> dict[str, Any]

Get all available fields for a model.

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

async def list_fields(client: AsyncOdooClient, model: str) -> dict[str, Any]:
    """Get all available fields for a model."""
    return await client.fields_get(model)

set_record_fields async

set_record_fields(client: AsyncOdooClient, model: str, record_id: int, values: dict[str, Any]) -> bool

Update fields on a record.

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

async def set_record_fields(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    values: dict[str, Any],
) -> bool:
    """Update fields on a record."""
    return await client.write(model, [record_id], values)

add_comment async

add_comment(client: AsyncOdooClient, model: str, record_id: int, message: str, user_id: int | None = None, markdown: bool = True) -> bool

Add a comment, requesting cross-user authorship only for internal users.

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

async def add_comment(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    message: str,
    user_id: int | None = None,
    markdown: bool = True,
) -> bool:
    """Add a comment, requesting cross-user authorship only for internal users."""
    body = _convert_to_html(message, markdown)
    return await message_post_sudo(
        client,
        model,
        record_id,
        body,
        user_id=user_id,
        is_note=False,
    )

add_note async

add_note(client: AsyncOdooClient, model: str, record_id: int, message: str, user_id: int | None = None, markdown: bool = True) -> bool

Add a note, requesting cross-user authorship only for internal users.

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

async def add_note(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    message: str,
    user_id: int | None = None,
    markdown: bool = True,
) -> bool:
    """Add a note, requesting cross-user authorship only for internal users."""
    body = _convert_to_html(message, markdown)
    return await message_post_sudo(
        client,
        model,
        record_id,
        body,
        user_id=user_id,
        is_note=True,
    )

list_tags async

list_tags(client: AsyncOdooClient, model: str) -> list[dict[str, Any]]

List available tags for a model.

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

async def list_tags(client: AsyncOdooClient, model: str) -> list[dict[str, Any]]:
    """List available tags for a model."""
    fields = _TAG_FIELDS
    return await client.search_read(model, fields=fields, order="name")

add_tag_to_record async

add_tag_to_record(client: AsyncOdooClient, model: str, record_id: int, tag_id: int) -> bool

Add a tag to a record.

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

async def add_tag_to_record(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    tag_id: int,
) -> bool:
    """Add a tag to a record."""
    record = await get_record(client, model, record_id, fields=["tag_ids"])
    current_tags = record.get("tag_ids", [])
    if tag_id not in current_tags:
        current_tags.append(tag_id)
        return await client.write(
            model,
            [record_id],
            {"tag_ids": [(6, 0, current_tags)]},
        )
    return True

list_messages async

list_messages(client: AsyncOdooClient, model: str, record_id: int, limit: int | None = None) -> list[dict[str, Any]]

List messages/chatter for a record.

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

async def list_messages(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """List messages/chatter for a record."""
    domain: list[Any] = [
        ("model", "=", model),
        ("res_id", "=", record_id),
    ]
    fields = _MESSAGE_FIELDS
    return await client.search_read(
        "mail.message",
        domain=domain,
        fields=fields,
        order="date desc",
        limit=limit,
    )

list_attachments async

list_attachments(client: AsyncOdooClient, model: str, record_id: int) -> list[dict[str, Any]]

List attachments for a record.

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

async def list_attachments(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
) -> list[dict[str, Any]]:
    """List attachments for a record."""
    domain: list[Any] = [
        ("res_model", "=", model),
        ("res_id", "=", record_id),
    ]
    fields = _ATTACHMENT_LIST_FIELDS
    return await client.search_read("ir.attachment", domain=domain, fields=fields)

download_attachment async

download_attachment(client: AsyncOdooClient, attachment_id: int, output_path: Path | None = None) -> Path

Download an attachment.

PARAMETER	DESCRIPTION
client	Async Odoo client TYPE: AsyncOdooClient
attachment_id	Attachment ID TYPE: int
output_path	Output file path TYPE: Path | None DEFAULT: None

RETURNS	DESCRIPTION
Path	Path to downloaded file

RAISES	DESCRIPTION
RecordNotFoundError	If attachment not found

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

async def download_attachment(
    client: AsyncOdooClient,
    attachment_id: int,
    output_path: Path | None = None,
) -> Path:
    """Download an attachment.

    Args:
        client: Async Odoo client
        attachment_id: Attachment ID
        output_path: Output file path

    Returns:
        Path to downloaded file

    Raises:
        RecordNotFoundError: If attachment not found
    """
    attachments = await client.read("ir.attachment", [attachment_id], _ATTACHMENT_READ_FIELDS)
    if not attachments:
        raise RecordNotFoundError("ir.attachment", attachment_id)

    attachment = attachments[0]
    filename = attachment.get("name", f"attachment_{attachment_id}")

    if output_path is None:
        output_path = Path.cwd() / filename
    elif output_path.is_dir():
        output_path = output_path / filename

    if attachment.get("datas"):
        data = base64.b64decode(attachment["datas"])
        output_path.write_bytes(data)
    else:
        raise RecordNotFoundError("ir.attachment", attachment_id)

    return output_path

download_record_attachments async

download_record_attachments(client: AsyncOdooClient, model: str, record_id: int, output_dir: Path | None = None, extension: str | None = None) -> list[Path]

Download all attachments for a record.

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

async def download_record_attachments(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    output_dir: Path | None = None,
    extension: str | None = None,
) -> list[Path]:
    """Download all attachments for a record."""
    if output_dir is None:
        output_dir = Path.cwd()
    elif not output_dir.exists():
        output_dir.mkdir(parents=True, exist_ok=True)

    attachments = await list_attachments(client, model, record_id)

    if extension:
        ext = extension.lower().lstrip(".")
        attachments = [
            att for att in attachments if att.get("name", "").lower().endswith(f".{ext}")
        ]

    downloaded_files: list[Path] = []

    for attachment in attachments:
        filename = attachment.get("name", f"attachment_{attachment['id']}")
        try:
            att_data = await client.read(
                "ir.attachment", [attachment["id"]], _ATTACHMENT_READ_FIELDS
            )
            if not att_data:
                continue

            att = att_data[0]
            filename = att.get("name", f"attachment_{attachment['id']}")
            output_path = output_dir / filename

            if att.get("datas"):
                data = base64.b64decode(att["datas"])
                output_path.write_bytes(data)
                downloaded_files.append(output_path)
        except Exception as e:
            import logging

            logging.getLogger("vodoo").warning("Failed to download %s: %s", filename, e)
            continue

    return downloaded_files

get_attachment_data async

get_attachment_data(client: AsyncOdooClient, attachment_id: int) -> bytes

Read an attachment and return its raw binary content in-memory.

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

async def get_attachment_data(
    client: AsyncOdooClient,
    attachment_id: int,
) -> bytes:
    """Read an attachment and return its raw binary content in-memory."""
    attachments = await client.read("ir.attachment", [attachment_id], _ATTACHMENT_READ_FIELDS)

    if not attachments:
        raise RecordNotFoundError("ir.attachment", attachment_id)

    return _decode_attachment_data(attachments[0], attachment_id)

get_record_attachment_data async

get_record_attachment_data(client: AsyncOdooClient, model: str, record_id: int) -> list[tuple[int, str, bytes]]

Read all attachments for a record and return their binary content in-memory.

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

async def get_record_attachment_data(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
) -> list[tuple[int, str, bytes]]:
    """Read all attachments for a record and return their binary content in-memory."""
    attachments = await list_attachments(client, model, record_id)

    result: list[tuple[int, str, bytes]] = []
    for att_meta in attachments:
        att_id = att_meta["id"]
        try:
            att_data = await client.read(
                "ir.attachment", [att_id], ["id", *_ATTACHMENT_READ_FIELDS]
            )
            if not att_data:
                continue
            decoded = _decode_attachment_record(att_data[0], att_id)
            if decoded is not None:
                result.append(decoded)
        except Exception as e:
            import logging

            logging.getLogger("vodoo").warning("Failed to read attachment %s: %s", att_id, e)
            continue

    return result

create_attachment async

create_attachment(client: AsyncOdooClient, model: str, record_id: int, file_path: Path | str | None = None, *, data: bytes | None = None, name: str | None = None) -> int

Create an attachment for a record.

PARAMETER	DESCRIPTION
client	Async Odoo client TYPE: AsyncOdooClient
model	Model name TYPE: str
record_id	Record ID TYPE: int
file_path	Path to file to attach (mutually exclusive with data) TYPE: Path | str | None DEFAULT: None
data	Raw bytes to attach (mutually exclusive with file_path) TYPE: bytes | None DEFAULT: None
name	Attachment name (defaults to filename; required when using data) TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
int	ID of created attachment

RAISES	DESCRIPTION
FileNotFoundError	If file path is invalid
ValueError	If arguments are invalid

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

async def create_attachment(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    file_path: Path | str | None = None,
    *,
    data: bytes | None = None,
    name: str | None = None,
) -> int:
    """Create an attachment for a record.

    Args:
        client: Async Odoo client
        model: Model name
        record_id: Record ID
        file_path: Path to file to attach (mutually exclusive with data)
        data: Raw bytes to attach (mutually exclusive with file_path)
        name: Attachment name (defaults to filename; required when using data)

    Returns:
        ID of created attachment

    Raises:
        FileNotFoundError: If file path is invalid
        ValueError: If arguments are invalid
    """
    values = _prepare_attachment_upload(file_path, data, name, model, record_id)
    return await client.create("ir.attachment", values)

parse_field_assignment async

parse_field_assignment(client: AsyncOdooClient, model: str, record_id: int, field_assignment: str, no_markdown: bool = False) -> tuple[str, Any]

Parse a field assignment and return field name and computed value. HTML fields automatically get markdown conversion unless no_markdown=True.

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

async def parse_field_assignment(
    client: AsyncOdooClient,
    model: str,
    record_id: int,
    field_assignment: str,
    no_markdown: bool = False,
) -> tuple[str, Any]:
    """Parse a field assignment and return field name and computed value.
    HTML fields automatically get markdown conversion unless no_markdown=True.
    """
    field, operator, value = _match_field_assignment(field_assignment)
    parsed_value = _parse_raw_value(field, value)

    # Auto-convert markdown to HTML for HTML fields
    if isinstance(parsed_value, str) and not no_markdown:
        fields_info = await list_fields(client, model)
        if field in fields_info and fields_info[field].get("type") == "html":
            parsed_value = _convert_to_html(parsed_value, use_markdown=True)
    # Handle operators that require current value
    if operator in ("+=", "-=", "*=", "/="):
        record = await get_record(client, model, record_id, fields=[field])
        current_value = record.get(field)
        parsed_value = _apply_operator(field, operator, parsed_value, current_value)

    return field, parsed_value