from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class PushNotificationTargetsRow(RepresentationRowBase):
PushTargetID: IntElement = "PushTargetID"
UserID: IntElement = "UserID"
Nickname: StringElement = "Nickname"
GcmToken: StringElement = "GcmToken"
QtwoToken: StringElement = "QtwoToken"
[docs]
class PushNotificationTargets(DbObject):
"""
Used to find the specific device target string to do a push notification
"""
NAME = "PushNotificationTargets"
REPRESENTATION_ROW_CLASS = PushNotificationTargetsRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_push_notification_targets")
subparser.set_defaults(parser="get")
subparser.add_argument("user_id", type=int, help="Q2_User.UserID")
subparser.add_argument(
"-d", "--disabled", type=bool, help="--show disabled users"
)
subparser.add_argument(
"-del", "--deleted", type=bool, help="--show deleted users"
)
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
[docs]
async def get(
self,
user_id: int,
disabled: bool = False,
deleted: bool = False,
serialize_for_cli=False,
) -> List[PushNotificationTargetsRow]:
assert isinstance(user_id, int), "Please supply a valid user id"
response = await self.call_hq(
"sdk_GetPushNotificationTargets",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "user_id", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"include_disabled",
disabled,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit, "include_deleted", deleted
),
]),
)
if serialize_for_cli:
columns = [
"PushTargetID",
"UserID",
"Nickname",
"GcmToken",
"QtwoToken",
"Created",
"Enabled",
"StatusChanged",
"DeletedAt",
]
response = self.serialize_for_cli(response, columns)
return response