from typing import Unpack
from q2_sdk.core.install_steps.base import (
InstallStep,
InstallStepAttribute,
InstallStepArguments,
)
from q2_sdk.core.exceptions import DatabaseDataError
from q2_sdk.hq.db.host_tran_code_group import (
HostTranCodeGroup as HostTranCodeGroupDbObj,
)
[docs]
class HostTranCodeGroup(InstallStep):
"""Interacts with the Q2_HostTranCodeGroup table"""
def __init__(
self,
description: str,
**kwargs: Unpack[InstallStepArguments],
):
super().__init__(**kwargs)
self.description = InstallStepAttribute(description)
[docs]
async def install(self):
await super().install()
db_obj = HostTranCodeGroupDbObj(self.logger, hq_credentials=self.hq_credentials)
already_exists = False
try:
await db_obj.get_by_name(self.description.value)
already_exists = True
except DatabaseDataError:
pass
if already_exists:
return
await db_obj.create(description=self.description.value)
[docs]
async def uninstall(self):
db_obj = HostTranCodeGroupDbObj(self.logger, self.hq_credentials)
await db_obj.delete(self.description.value)