# This is an autogenerated file from the command "q2 generate_hq_api" and will be overwritten if run again
from typing import cast, Optional
from q2_sdk.core.q2_logging.logger import Q2LoggerType
from q2_sdk.hq.models.hq_credentials import BackOfficeHqCredentials
from q2_sdk.hq.models.hq_params.backoffice import BackOfficeAuthenticatedParamsObj
from q2_sdk.hq.models.hq_request.backoffice import BackOfficeRequest
from q2_sdk.hq.models.backoffice_response import BackOfficeResponse
from q2_sdk.hq.models.dataset_builder import DataSetBuilder
from q2_sdk.hq.hq_api.backoffice import GetMessageGroupMembers
[docs]
class ParamsObj(BackOfficeAuthenticatedParamsObj):
"""Parameters definition for SetMessageGroupMembers
.. seealso:: :mod:`~q2_sdk.hq.hq_api.backoffice.SetMessageGroupMembers_models` for DataSet schemas and builder functions.
"""
def __init__(
self,
logger: Q2LoggerType,
hq_credentials: BackOfficeHqCredentials,
message_group_member_edit: Optional[DataSetBuilder] = None,
):
"""
:param logger: Reference to calling request's logger (self.logger in your extension)
:param message_group_member_edit: DataSetBuilder instance
:param hq_credentials: Defaults to settings.HQ_CREDENTIALS
"""
super().__init__(logger, hq_credentials)
self.request_params.update({
"messageGroupMemberEdit": message_group_member_edit,
})
[docs]
async def execute(params_obj: ParamsObj, use_json=True, **kwargs) -> BackOfficeResponse:
"""
This is the default way to submit the request to HQ.
This endpoint uses DataSet parameters and defaults to the JSON (.ashx)
transport. SOAP (.asmx) is not supported for DataSet serialization.
:param params_obj: Object containing everything necessary to call this HQ endpoint
:param use_json: If True, will call HQ's .ashx (json) endpoint instead of .asmx (soap)
"""
request = BackOfficeRequest("SetMessageGroupMembers", use_json=use_json, **kwargs)
return cast(BackOfficeResponse, await request.execute(params_obj, **kwargs))
[docs]
def build_json(params_obj: ParamsObj):
return BackOfficeRequest.build_json(params_obj)
[docs]
async def fetch_current(
logger: Q2LoggerType,
hq_credentials: BackOfficeHqCredentials,
) -> dict[str, list]:
"""
Fetch current state from GetMessageGroupMembers for optimistic concurrency.
Calls the GET operation and parses the response into typed model instances
with ``original_record`` pre-populated. Modify the returned instances and
pass them to the corresponding ``build_*_dataset()`` function.
Workflow::
# 1. Fetch current state (pre-populates original_record on each instance)
current = await SetMessageGroupMembers.fetch_current(logger, hq_creds, ...)
# 2. Modify the returned dataclass instances
instance = current["ModelClassName"][0]
instance.some_field = new_value
# 3. Build dataset and execute (original_record produces the Unchanged row)
dataset = build_*_dataset(instance)
params = SetMessageGroupMembers.ParamsObj(logger, hq_creds, param=dataset)
await SetMessageGroupMembers.execute(params)
.. seealso:: :mod:`~q2_sdk.hq.hq_api.backoffice.SetMessageGroupMembers_models`
:param logger: Calling request's logger (self.logger in your extension)
:param hq_credentials: HQ credentials for the BackOffice session
:returns: Dict mapping model class names to lists of populated instances
:raises RuntimeError: If the GET operation fails or returns no data
"""
# The models module imports this wrapper, so import it lazily to avoid a
# circular import at module load. The paired GET module is imported at module
# scope (see the header) so its enum-typed params resolve in the signature.
from q2_sdk.hq.hq_api.backoffice.SetMessageGroupMembers_models import (
Q2MessageGroupMemberEdit,
)
params = GetMessageGroupMembers.ParamsObj(logger, hq_credentials)
response = await GetMessageGroupMembers.execute(params, use_json=False)
if not response.success or response.result_node is None:
raise RuntimeError(
f"GetMessageGroupMembers failed: {response.error_message or 'no result'}"
)
result: dict[str, list] = {}
for row_elem in response.result_node.findall(".//Q2_MessageGroupMemberEdit"):
instance = Q2MessageGroupMemberEdit.from_hq_response(row_elem)
result.setdefault("Q2MessageGroupMemberEdit", []).append(instance)
return result