Source code for q2_sdk.hq.db.ui_source

from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement, BoolElement
from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dev_only
from q2_sdk.tools.utils import to_bool
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase




class UiSourceRow(RepresentationRowBase):
    # object name: type hinting = "column name in the db response"
    UISourceID: IntElement = "UISourceID"
    ShortName: StringElement = "ShortName"
    PasswordPolicyID: IntElement = "PasswordPolicyID"
    Enabled: BoolElement = "Enabled"
    EnableAuthToken: BoolElement = "EnableAuthToken"
    EnableAuthAccessCode: BoolElement = "EnableAuthAccessCode"





class UiSource(DbObject):
    # GET_BY_NAME_KEY = "column in the db response"
    NAME = "UiSource"
    REPRESENTATION_ROW_CLASS = UiSourceRow



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

        subparser = parser.add_parser("update_ui_source")
        subparser.set_defaults(parser="update")
        subparser.set_defaults(func=partial(self.update))
        subparser.add_argument("short_name", help="Q2_UiSource.ShortName")
        subparser.add_argument(
            "enable_auth_token", type=to_bool, help="Q2_UiSource.EnableAuthToken"
        )
        subparser.add_argument(
            "enable_auth_access_code",
            type=to_bool,
            help="Q2_UiSource.EnableAuthAccessCode",
        )




    async def get(self, serialize_for_cli=False) -> List[UiSourceRow]:
        response = await self.call_hq(
            "sdk_GetUiSource",
        )

        if serialize_for_cli:
            response = self.serialize_for_cli(response)

        return response




    @dev_only
    async def update(
        self,
        short_name: str,
        enable_auth_access_code: bool,
        enable_auth_token: bool,
    ):
        response = await self.call_hq(
            "sdk_UpdateUiSource",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "short_name",
                    short_name,
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Bit,
                    "enable_auth_token",
                    enable_auth_token,
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Bit,
                    "enable_auth_access_code",
                    enable_auth_access_code,
                ),
            ]),
        )
        return response