Source code for q2_sdk.hq.db.marketing_page_name

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 MarketingPageNameRow(TableRow): MarketingPageNameID: int ShortName: str Description: str AllowMarketing: bool SystemNotification: bool WedgeAddressID: int
[docs] class SystemNotificationRow(TableRow): CampaignName: str StartDate: str EndDate: str ZoneID: int
[docs] class MarketingPageName(DbObject): GET_BY_NAME_KEY = "ShortName" NAME = "MarketingPageName" REPRESENTATION_ROW_CLASS = MarketingPageNameRow
[docs] def add_arguments(self, parser: _SubParsersAction): subparser = parser.add_parser("get_marketing_page_name") subparser.set_defaults(parser="get") subparser.set_defaults(func=partial(self.get, serialize_for_cli=True)) subparser.add_argument("short_name", help="Q2_MarketingPageName.ShortName") subparser = parser.add_parser("get_system_notifications") subparser.set_defaults(parser="get_system_notifications") subparser.set_defaults( func=partial(self.get_system_notifications, serialize_for_cli=True) ) subparser = parser.add_parser("add_marketing_page_name") subparser.set_defaults(parser="add") subparser.set_defaults(func=partial(self.add)) subparser.add_argument("short_name", help="Q2_MarketingPageName.ShortName") subparser.add_argument("description", help="Q2_MarketingPageName.Description") subparser.add_argument( "allow_marketing", help="Q2_MarketingPageName.AllowMarketing" ) subparser.add_argument( "system_notification", help="Q2_MarketingPageName.SystemNotification" ) subparser.add_argument( "-wa", "--wedge-address-id", help="Q2_MarketingPageName.WedgeAddressID" )
[docs] async def get( self, short_name: str, serialize_for_cli=False ) -> List[MarketingPageNameRow]: response = await self.call_hq( "sdk_GetMarketingPageName", ExecuteStoredProcedure.SqlParameters([ ExecuteStoredProcedure.SqlParam( ExecuteStoredProcedure.DataType.VarChar, "short_name", short_name, ) ]), ) if serialize_for_cli: columns = [ "MarketingPageNameID", "ShortName", "Description", "AllowMarketing", "SystemNotification", "WedgeAddressID", ] response = self.serialize_for_cli(response, columns) return response
[docs] async def add( self, short_name: str, description: str, allow_marketing: bool, system_notification: bool, wedge_address_id: int = None, ): response = await self.call_hq( "sdk_AddMarketingPageName", ExecuteStoredProcedure.SqlParameters([ ExecuteStoredProcedure.SqlParam( ExecuteStoredProcedure.DataType.VarChar, "short_name", short_name, ), ExecuteStoredProcedure.SqlParam( ExecuteStoredProcedure.DataType.VarChar, "description", description, ), ExecuteStoredProcedure.SqlParam( ExecuteStoredProcedure.DataType.Bit, "allow_marketing", allow_marketing, ), ExecuteStoredProcedure.SqlParam( ExecuteStoredProcedure.DataType.Bit, "system_notification", system_notification, ), ExecuteStoredProcedure.SqlParam( ExecuteStoredProcedure.DataType.Int, "wedge_address_id", wedge_address_id, ), ]), ) return response
[docs] async def get_system_notifications( self, serialize_for_cli=False ) -> list[SystemNotificationRow]: """ Fetches Marketing messages that are marked as System Notifications and enabled. Can be used with Q2API GetMarketingCampaign call. """ response = await self.call_hq( "sdk_GetSystemNotifications", ExecuteStoredProcedure.SqlParameters([]) ) if serialize_for_cli: columns = ["CampaignName", "StartDate", "EndDate", "ZoneID"] response = self.serialize_for_cli(response, columns) return response