Source code for q2_sdk.hq.db.recurring_transaction

from argparse import _SubParsersAction
from functools import partial
from typing import List

from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.table_row import TableRow

from .db_object import DbObject


[docs] class RecurringTransactionRow(TableRow): RecurringTransactionID: int TransactionID: int CustomerID: int UserID: int TransactionType: str EveryXWeeksOrMonths: int NumberOfOccurrences: int StartDate: str EndDate: str RecurringTransactionType: str ExecutedOccurrences: int LastOccurrence: str
[docs] class RecurringTransaction(DbObject): # GET_BY_NAME_KEY = "column in the db response" NAME = "RecurringTransaction" REPRESENTATION_ROW_CLASS = RecurringTransactionRow
[docs] def add_arguments(self, parser: _SubParsersAction): subparser = parser.add_parser("get_recurring_transaction") subparser.set_defaults(parser="get") subparser.set_defaults(func=partial(self.get, serialize_for_cli=True)) subparser.add_argument( "recurring_transaction_id", help="Q2_RecurringTransaction.RecurringTransactionID", type=int, )
[docs] async def get( self, recurring_transaction_id: int, serialize_for_cli=False ) -> List[RecurringTransactionRow]: response = await self.call_hq( "sdk_GetRecurringTransaction", ExecuteStoredProcedure.SqlParameters([ ExecuteStoredProcedure.SqlParam( ExecuteStoredProcedure.DataType.Int, "recurring_transaction_id", recurring_transaction_id, ) ]), ) if serialize_for_cli: columns = [ "RecurringTransactionID", "TransactionID", "CustomerID", "UserID", "TransactionType", "EveryXWeeksOrMonths", "NumberOfOccurrences", "StartDate", "EndDate", "RecurringTransactionType", "ExecutedOccurrences", "LastOccurrence", ] response = self.serialize_for_cli( response, columns, fields_to_truncate=["StartDate", "EndDate"] ) return response