from dataclasses import dataclass, fields
from typing import Optional
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.core.exceptions import DatabaseDataError
from .db_object import DbObject
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
@dataclass
class UpdateParameters:
admin_login_name: str
first_name: Optional[str] = None
last_name: Optional[str] = None
email: Optional[str] = None
[docs]
def build_stored_proc_parameters(self) -> list:
parameters = []
for field in fields(self):
name = field.name
value = getattr(self, name)
if value:
Param(value, D_TYPES.VarChar, name).add_to_param_list(parameters)
return parameters
[docs]
class AdminUser(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "AdminUser"
[docs]
async def update_info(self, update_data: UpdateParameters):
if not any([update_data.email, update_data.first_name, update_data.last_name]):
raise DatabaseDataError("Nothing to update")
stored_proc_params = ExecuteStoredProcedure.SqlParameters(
update_data.build_stored_proc_parameters()
)
response = await self.call_hq("sdk_UpdateAdminUser", stored_proc_params)
return response
[docs]
async def get_all(
self, include_deleted: bool = False, serialize_for_cli: bool = False
):
pass