from argparse import _SubParsersAction
from functools import partial
from typing import Optional
from lxml.objectify import StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dev_only
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class ApiStoredProcRow(RepresentationRowBase):
StoredProcShortName: StringElement = "StoredProcShortName"
Description: StringElement = "Description"
StoredProcName: StringElement = "StoredProcName"
[docs]
class ApiStoredProc(DbObject):
"""
Allows for Access to the Q2_ApiStoredProc table, which registers stored procedures to use within the SDK
via ExecuteStoredProcedure calls
"""
REPRESENTATION_ROW_CLASS = ApiStoredProcRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_api_stored_procs")
subparser.set_defaults(parser="get_api_stored_procs")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("add_api_stored_proc")
subparser.set_defaults(parser="add_api_stored_proc")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument(
"short_name", help="Q2_ApiStoredProc.StoredProcShortName"
)
subparser.add_argument("audit_action", help="Q2_AuditAction.ShortName")
subparser = parser.add_parser("remove_api_stored_proc")
subparser.set_defaults(parser="remove_api_stored_proc")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"short_name", help="Q2_ApiStoredProc.StoredProcShortName"
)
[docs]
async def get(self, serialize_for_cli=False) -> list[ApiStoredProcRow]:
response = await self.call_hq("sdk_GetApiStoredProcs")
if serialize_for_cli:
response = self.serialize_for_cli(
response, ["StoredProcShortName", "Description", "StoredProcName"]
)
return response
[docs]
async def create(
self,
short_name,
description: Optional[str] = None,
name: Optional[str] = None,
audit_action: Optional[str] = None,
):
"""
Creates row in Q2_ApiStoredProc table
:param short_name: Unique name of the stored procedure
:param description: (Optional) defaults to name
:param name: (Optional) Defaults to short_name
:param audit_action: (Optional) Defaults to NULL
"""
if not name:
name = short_name
if not description:
description = name
return await self.call_hq(
"sdk_AddApiStoredProc",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"description",
description,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "name", name
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"audit_action",
audit_action,
),
]),
)
[docs]
@dev_only
async def delete(self, short_name):
"""Note: this only works in the dev environment"""
return await self.call_hq(
"sdk_RemoveApiStoredProc",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
)
]),
)