"""
Utility for building HQ DataSet payloads for BackOffice operations.
Row-state markers:
M = Modified — the new values you want the record to have
U = Unchanged — the original values before your edit (for concurrency check)
A = Added — a new record (typically uses negative IDs)
D = Deleted — a record to remove
For **update** operations, each modified record requires a pair of rows:
a Modified row (new values) and an Unchanged row (original values). HQ compares
the Unchanged row against the database to ensure no one else edited the record
since you last read it. Without the original values, HQ rejects with error -8.
Use ``fetch_current()`` on the endpoint wrapper to GET the current state, then
build your Modified row with the changes and the Unchanged row from the original.
For **add** operations, only an Added row is needed — no prior GET required.
A DataSet payload can contain multiple tables, each with its own columns and rows.
"""
from __future__ import annotations
from enum import Enum
from typing import Any
from q2_sdk.hq.models.hq_params.base import BaseParameter
[docs]
class RowState(Enum):
"""Row-state markers for HQ DataSet rows."""
MODIFIED = "M"
UNCHANGED = "U"
ADDED = "A"
DELETED = "D"
[docs]
class DataSetTable:
"""
Represents a single table within an HQ DataSet.
Usage::
table = DataSetTable("Q2_User", ["UserID", "FirstName", "LastName"])
# Update: change LastName from "Smith" to "Doe"
table.add_unchanged_row([1, "John", "Smith"]) # original values
table.add_modified_row([1, "John", "Doe"]) # new values
"""
def __init__(self, name: str, columns: list[str]):
self.name = name
self.columns = list(columns)
self.rows: list[dict[str, list]] = []
[docs]
def add_row(self, state: RowState, values: list[Any]) -> "DataSetTable":
"""
Add a row with the given state marker and column values.
:param state: RowState enum indicating the row's change state
:param values: List of values corresponding to self.columns (positional)
:returns: self for chaining
"""
if len(values) != len(self.columns):
raise ValueError(
f"Table '{self.name}': expected {len(self.columns)} values "
f"({self.columns}), got {len(values)}"
)
self.rows.append({state.value: values})
return self
[docs]
def add_modified_row(self, values: list[Any]) -> "DataSetTable":
"""Shortcut for add_row(RowState.MODIFIED, values)."""
return self.add_row(RowState.MODIFIED, values)
[docs]
def add_unchanged_row(self, values: list[Any]) -> "DataSetTable":
"""Shortcut for add_row(RowState.UNCHANGED, values)."""
return self.add_row(RowState.UNCHANGED, values)
[docs]
def add_added_row(self, values: list[Any]) -> "DataSetTable":
"""Shortcut for add_row(RowState.ADDED, values)."""
return self.add_row(RowState.ADDED, values)
[docs]
def add_deleted_row(self, values: list[Any]) -> "DataSetTable":
"""Shortcut for add_row(RowState.DELETED, values)."""
return self.add_row(RowState.DELETED, values)
[docs]
def serialize(self) -> dict:
"""Serialize this table to the HQ JSON format."""
return {"columns": self.columns, "rows": self.rows}
[docs]
class DataSetBuilder(BaseParameter):
"""
Builds HQ DataSet payloads containing one or more tables.
Passed directly as a parameter value to endpoint ParamsObj instances.
For most update operations, prefer the generated ``_models.py`` helpers
(e.g., ``UpdateGroup_models.build_group_edit_dataset()``) which handle
row construction from typed dataclasses. Use DataSetBuilder directly only
when building payloads manually.
Usage::
dataset = DataSetBuilder("DalGroupEdit")
dataset.add_table("Q2_Group", ["GroupID", "GroupDesc"])
dataset.table("Q2_Group").add_unchanged_row([1, "Old Name"]) # original values
dataset.table("Q2_Group").add_modified_row([1, "New Name"]) # new values
params = UpdateGroup.ParamsObj(logger, hq_creds, group_edit=dataset)
await UpdateGroup.execute(params)
"""
def __init__(self, dataset_name: str = "NewDataSet"):
self.dataset_name = dataset_name
self._tables: dict[str, DataSetTable] = {}
[docs]
def add_table(self, name: str, columns: list[str]) -> DataSetTable:
"""
Add a new table to this dataset.
:param name: The table name (e.g., "Q2_User", "Q2_Email")
:param columns: List of column names for this table
:returns: The created DataSetTable instance
"""
tbl = DataSetTable(name, columns)
self._tables[name] = tbl
return tbl
[docs]
def table(self, name: str) -> DataSetTable:
"""
Get an existing table by name.
:param name: The table name
:returns: The DataSetTable instance
:raises KeyError: If the table doesn't exist
"""
if name not in self._tables:
raise KeyError(
f"Table '{name}' not found in dataset '{self.dataset_name}'. "
f"Available tables: {list(self._tables.keys())}"
)
return self._tables[name]
[docs]
def serialize_as_json(self) -> dict:
"""
Serialize the entire dataset to HQ's expected JSON format.
Returns a dict like::
{
"$type": "DataSet",
"Q2_User": {"columns": [...], "rows": [...]},
"Q2_Email": {"columns": [...], "rows": [...]}
}
"""
result = {"$type": "DataSet"}
for name, tbl in self._tables.items():
result[name] = tbl.serialize()
return result
[docs]
def serialize_as_xml(self):
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported. "
"HQ's .asmx endpoint requires ADO.NET's DiffGram format: inline XSD schema "
"defining every column type, a <diffgr:before> section with original row values "
"for concurrency, row identity attributes (diffgr:id, msdata:rowOrder, "
"diffgr:hasChanges), and multiple coordinated XML namespaces. "
"Use JSON mode (use_json=True) for operations with dataset parameters."
)