from typing import Unpack
from q2_sdk.core.install_steps.base import (
InstallStep,
InstallStepAttribute,
InstallStepArguments,
)
from q2_sdk.hq.db.report_info import ReportInfo as ReportInfoDbObj
[docs]
class ReportInfo(InstallStep):
"""Interacts with the Q2_ReportInfo table"""
def __init__(
self,
title,
category,
name,
info_obj,
wedge_address_typename,
**kwargs: Unpack[InstallStepArguments],
):
super().__init__(**kwargs)
self.title = InstallStepAttribute(title)
self.category = InstallStepAttribute(category)
self.name = InstallStepAttribute(name)
self.info_obj = InstallStepAttribute(info_obj)
self.wedge_address_typename = InstallStepAttribute(wedge_address_typename)
[docs]
async def install(self):
await super().install()
report_info_obj = ReportInfoDbObj(
self.logger, hq_credentials=self.hq_credentials
)
already_exists = False
all_reports = await report_info_obj.get()
for report in all_reports:
if report.findtext("ReportTitle") == self.title.value:
already_exists = True
break
if already_exists:
desired_func = report_info_obj.update
else:
desired_func = report_info_obj.create
await desired_func(
self.title.value,
self.category.value,
self.name.value,
self.info_obj.value,
self.wedge_address_typename.value,
)
[docs]
async def uninstall(self):
await ReportInfoDbObj(self.logger, hq_credentials=self.hq_credentials).delete(
self.title.value
)