# 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 lxml import etree
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.base import BaseParameter
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
[docs]
class transactions(BaseParameter):
def __init__(
self,
batch_id: int,
processing_start_time: str,
admin_user_id: int,
zone_id: Optional[int] = None,
batch_description: Optional[str] = None,
generated_transactions: Optional[DataSetBuilder] = None,
change_of_address: Optional[DataSetBuilder] = None,
check_reorder: Optional[DataSetBuilder] = None,
funds_transfer: Optional[DataSetBuilder] = None,
stop_payment: Optional[DataSetBuilder] = None,
bill_payment: Optional[DataSetBuilder] = None,
ach_payment: Optional[DataSetBuilder] = None,
ach_iat_payment: Optional[DataSetBuilder] = None,
ach_receipt: Optional[DataSetBuilder] = None,
eftps: Optional[DataSetBuilder] = None,
payroll: Optional[DataSetBuilder] = None,
wire_domestic: Optional[DataSetBuilder] = None,
ach_collections: Optional[DataSetBuilder] = None,
ach_batch: Optional[DataSetBuilder] = None,
ach_iat_batch: Optional[DataSetBuilder] = None,
external_transfer: Optional[DataSetBuilder] = None,
wire_international: Optional[DataSetBuilder] = None,
send_check: Optional[DataSetBuilder] = None,
ach_pass_thru: Optional[DataSetBuilder] = None,
rtp_credit_transfer: Optional[DataSetBuilder] = None,
rtp_request_for_payment: Optional[DataSetBuilder] = None,
):
self.zone_id = zone_id
self.batch_id = batch_id
self.processing_start_time = processing_start_time
self.admin_user_id = admin_user_id
self.batch_description = batch_description
self.generated_transactions = generated_transactions
self.change_of_address = change_of_address
self.check_reorder = check_reorder
self.funds_transfer = funds_transfer
self.stop_payment = stop_payment
self.bill_payment = bill_payment
self.ach_payment = ach_payment
self.ach_iat_payment = ach_iat_payment
self.ach_receipt = ach_receipt
self.eftps = eftps
self.payroll = payroll
self.wire_domestic = wire_domestic
self.ach_collections = ach_collections
self.ach_batch = ach_batch
self.ach_iat_batch = ach_iat_batch
self.external_transfer = external_transfer
self.wire_international = wire_international
self.send_check = send_check
self.ach_pass_thru = ach_pass_thru
self.rtp_credit_transfer = rtp_credit_transfer
self.rtp_request_for_payment = rtp_request_for_payment
[docs]
def serialize_as_xml(self):
root = etree.Element("transactions")
sub_elem = etree.SubElement(root, "ZoneId")
sub_elem.text = str(self.zone_id)
sub_elem = etree.SubElement(root, "BatchId")
sub_elem.text = str(self.batch_id)
sub_elem = etree.SubElement(root, "ProcessingStartTime")
sub_elem.text = str(self.processing_start_time)
sub_elem = etree.SubElement(root, "AdminUserId")
sub_elem.text = str(self.admin_user_id)
sub_elem = etree.SubElement(root, "BatchDescription")
sub_elem.text = str(self.batch_description)
if self.generated_transactions is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'GeneratedTransactions'. Call execute() with the default use_json=True."
)
if self.change_of_address is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'ChangeOfAddress'. Call execute() with the default use_json=True."
)
if self.check_reorder is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'CheckReorder'. Call execute() with the default use_json=True."
)
if self.funds_transfer is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'FundsTransfer'. Call execute() with the default use_json=True."
)
if self.stop_payment is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'StopPayment'. Call execute() with the default use_json=True."
)
if self.bill_payment is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'BillPayment'. Call execute() with the default use_json=True."
)
if self.ach_payment is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'AchPayment'. Call execute() with the default use_json=True."
)
if self.ach_iat_payment is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'AchIatPayment'. Call execute() with the default use_json=True."
)
if self.ach_receipt is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'AchReceipt'. Call execute() with the default use_json=True."
)
if self.eftps is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'Eftps'. Call execute() with the default use_json=True."
)
if self.payroll is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'Payroll'. Call execute() with the default use_json=True."
)
if self.wire_domestic is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'WireDomestic'. Call execute() with the default use_json=True."
)
if self.ach_collections is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'AchCollections'. Call execute() with the default use_json=True."
)
if self.ach_batch is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'AchBatch'. Call execute() with the default use_json=True."
)
if self.ach_iat_batch is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'AchIatBatch'. Call execute() with the default use_json=True."
)
if self.external_transfer is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'ExternalTransfer'. Call execute() with the default use_json=True."
)
if self.wire_international is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'WireInternational'. Call execute() with the default use_json=True."
)
if self.send_check is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'SendCheck'. Call execute() with the default use_json=True."
)
if self.ach_pass_thru is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'AchPassThru'. Call execute() with the default use_json=True."
)
if self.rtp_credit_transfer is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'RtpCreditTransfer'. Call execute() with the default use_json=True."
)
if self.rtp_request_for_payment is not None:
raise NotImplementedError(
"DataSet serialization to SOAP XML is not supported for "
"'RtpRequestForPayment'. Call execute() with the default use_json=True."
)
return root
[docs]
def serialize_as_json(self):
return {
"ZoneId": self.zone_id,
"BatchId": self.batch_id,
"ProcessingStartTime": self.processing_start_time,
"AdminUserId": self.admin_user_id,
"BatchDescription": self.batch_description,
"GeneratedTransactions": self.generated_transactions.serialize_as_json()
if self.generated_transactions is not None
else None,
"ChangeOfAddress": self.change_of_address.serialize_as_json()
if self.change_of_address is not None
else None,
"CheckReorder": self.check_reorder.serialize_as_json()
if self.check_reorder is not None
else None,
"FundsTransfer": self.funds_transfer.serialize_as_json()
if self.funds_transfer is not None
else None,
"StopPayment": self.stop_payment.serialize_as_json()
if self.stop_payment is not None
else None,
"BillPayment": self.bill_payment.serialize_as_json()
if self.bill_payment is not None
else None,
"AchPayment": self.ach_payment.serialize_as_json()
if self.ach_payment is not None
else None,
"AchIatPayment": self.ach_iat_payment.serialize_as_json()
if self.ach_iat_payment is not None
else None,
"AchReceipt": self.ach_receipt.serialize_as_json()
if self.ach_receipt is not None
else None,
"Eftps": self.eftps.serialize_as_json() if self.eftps is not None else None,
"Payroll": self.payroll.serialize_as_json()
if self.payroll is not None
else None,
"WireDomestic": self.wire_domestic.serialize_as_json()
if self.wire_domestic is not None
else None,
"AchCollections": self.ach_collections.serialize_as_json()
if self.ach_collections is not None
else None,
"AchBatch": self.ach_batch.serialize_as_json()
if self.ach_batch is not None
else None,
"AchIatBatch": self.ach_iat_batch.serialize_as_json()
if self.ach_iat_batch is not None
else None,
"ExternalTransfer": self.external_transfer.serialize_as_json()
if self.external_transfer is not None
else None,
"WireInternational": self.wire_international.serialize_as_json()
if self.wire_international is not None
else None,
"SendCheck": self.send_check.serialize_as_json()
if self.send_check is not None
else None,
"AchPassThru": self.ach_pass_thru.serialize_as_json()
if self.ach_pass_thru is not None
else None,
"RtpCreditTransfer": self.rtp_credit_transfer.serialize_as_json()
if self.rtp_credit_transfer is not None
else None,
"RtpRequestForPayment": self.rtp_request_for_payment.serialize_as_json()
if self.rtp_request_for_payment is not None
else None,
}
[docs]
class ParamsObj(BackOfficeAuthenticatedParamsObj):
"""Parameters definition for ProcessGeneratedTransactions"""
def __init__(
self,
logger: Q2LoggerType,
hq_credentials: BackOfficeHqCredentials,
transactions: Optional[transactions] = None,
zone_id: Optional[int] = None,
):
"""
:param logger: Reference to calling request's logger (self.logger in your extension)
:param transactions:
:param zone_id:
:param hq_credentials: Defaults to settings.HQ_CREDENTIALS
"""
super().__init__(logger, hq_credentials)
self.request_params.update({
"transactions": transactions,
"zoneId": zone_id,
})
[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(
"ProcessGeneratedTransactions", 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)