Source code for q2_sdk.hq.db.form_to_group

import typing
from argparse import _SubParsersAction
from functools import partial
from q2_sdk.hq.db.user_property_data import UserPropertyData
from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject

PROPERTY_NAME = "Customer/CanSubmitForms"



[docs]
class FormToGroup(DbObject):

[docs]
    def add_arguments(self, parser: _SubParsersAction):
        subparser = parser.add_parser("get_form_to_group")
        subparser.set_defaults(parser="get_form_to_group")
        subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
        subparser.add_argument(
            "-g", "--group_id", help="Limit results to a single group"
        )
        subparser.add_argument("-f", "--form_id", help="Limit results to a single form")

        subparser = parser.add_parser("add_form_to_group")
        subparser.set_defaults(parser="add_form_to_group")
        subparser.set_defaults(func=partial(self.add))
        subparser.add_argument("form_id", help="Q2_Forms.FormID")
        subparser.add_argument("group_id", help="Q2_Groups.GroupID")

        subparser = parser.add_parser("remove_form_from_group")
        subparser.set_defaults(parser="remove_form_from_group")
        subparser.set_defaults(func=partial(self.delete))
        subparser.add_argument("form_id", help="Q2_Forms.FormID")
        subparser.add_argument("group_id", help="Q2_Groups.GroupID")



[docs]
    async def get(
        self,
        serialize_for_cli=False,
        form_id: typing.Optional[int] = None,
        group_id: typing.Optional[int] = None,
    ):
        response = await self.call_hq("sdk_GetFormToGroup")
        if form_id:
            form_id = int(form_id)
            response = [x for x in response if x.find("FormID") == form_id]
        if group_id:
            group_id = int(group_id)
            response = [x for x in response if x.find("GroupID") == group_id]
        if serialize_for_cli:
            columns = ["FormID", "GroupID", "ShortName", "GroupDesc"]
            response = self.serialize_for_cli(response, columns)
        return response



[docs]
    async def add(self, form_id: int, group_id: int):
        await self._enable_upd_for_group(group_id)
        return await self.call_hq(
            "sdk_AddFormToGroup",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "form_id", form_id
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "group_id", group_id
                ),
            ]),
        )



[docs]
    async def add_central(self, form_id: int, group_id: int):
        return await self.call_hq(
            "sdk_AddCentralFormToGroup",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "form_id", form_id
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "group_id", group_id
                ),
            ]),
        )



[docs]
    async def delete(self, form_id: int, group_id: int):
        return await self.call_hq(
            "sdk_RemoveFormFromGroup",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "form_id", form_id
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "group_id", group_id
                ),
            ]),
        )



[docs]
    async def delete_central(self, form_id: int, group_id: int):
        return await self.call_hq(
            "sdk_RemoveCentralFormFromGroup",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "form_id", form_id
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int, "group_id", group_id
                ),
            ]),
        )


    async def _enable_upd_for_group(self, group_id):
        user_property_data_obj = UserPropertyData(
            self.logger, hq_credentials=self.hq_credentials
        )
        existing = await user_property_data_obj.get(
            PROPERTY_NAME, group_id=group_id, strict=True
        )
        create = True
        does_exist = [
            x
            for x in existing
            if x.findtext("GroupID") and x.GroupID.text == str(group_id)
        ]
        if does_exist:
            create = False
        if create:
            await user_property_data_obj.create(PROPERTY_NAME, True, group_id=group_id)
        else:
            if does_exist[0].PropertyValue.text.lower() != "true":
                await user_property_data_obj.update(
                    PROPERTY_NAME, True, group_id=group_id
                )