Source code for q2_sdk.hq.db.report_info

from argparse import _SubParsersAction
from functools import partial

from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dev_only

from .db_object import DbObject




class ReportInfo(DbObject):
    GET_BY_NAME_KEY = "Name"
    NAME = "ReportInfo"



    def add_arguments(self, parser: _SubParsersAction):
        subparser = parser.add_parser("get_report_info")
        subparser.set_defaults(parser="get_report_info")
        subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))

        add_subparser = parser.add_parser("add_report_info")
        add_subparser.set_defaults(parser="add_report_info")
        add_subparser.set_defaults(func=partial(self.create))

        update_subparser = parser.add_parser("update_report_info")
        update_subparser.set_defaults(parser="update_report_info")
        update_subparser.set_defaults(func=partial(self.update))

        delete_subparser = parser.add_parser("remove_report_info")
        delete_subparser.set_defaults(parser="remove_report_info")
        delete_subparser.set_defaults(func=partial(self.delete))
        delete_subparser.add_argument("title", help="Q2_ReportInfo.ReportTitle")

        for subparser in [add_subparser, update_subparser]:
            subparser.add_argument("title", help="Report title displayed in Central")
            subparser.add_argument("category", help="Report Category")
            subparser.add_argument("name", help="Report Name")
            subparser.add_argument("info_object", help="Report Info Object")
            subparser.add_argument(
                "wedge_address_typename",
                help=("Wedge address type name to set WedgeAddressIDin Q2_ReportInfo"),
            )




    async def get(self, serialize_for_cli=False):
        response = await self.call_hq("sdk_GetReportInfo")
        if serialize_for_cli:
            columns = [
                "ReportTitle",
                "Category",
                "Name",
                "InfoObject",
                "WedgeAddressTypeName",
            ]
            response = self.serialize_for_cli(response, columns)
        return response




    async def create(
        self,
        title: str,
        category: str,
        name: str,
        info_object: str,
        wedge_address_typename: str,
    ):
        return await self.call_hq(
            "sdk_AddReportInfo",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar, "reportTitle", title
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "reportCategory",
                    category,
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar, "reportName", name
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "reportInfoObject",
                    info_object,
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "wedgeAddressTypeName",
                    wedge_address_typename,
                ),
            ]),
        )




    @dev_only
    async def delete(self, title: str):
        """Note: this only works in the dev environment"""
        return await self.call_hq(
            "sdk_RemoveReportInfo",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar, "title", title
                )
            ]),
        )




    async def update(
        self,
        title: str,
        category: str,
        name: str,
        info_object: str,
        wedge_address_typename: str,
    ):
        return await self.call_hq(
            "sdk_UpdateReportInfo",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar, "reportTitle", title
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "reportCategory",
                    category,
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar, "reportName", name
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "reportInfoObject",
                    info_object,
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "wedgeAddressTypeName",
                    wedge_address_typename,
                ),
            ]),
        )