from dataclasses import dataclass
from typing import Optional
from q2_sdk.core import q2_requests
from q2_sdk.core.q2_logging.logger import Q2LoggerType
from q2_sdk.entrypoints import TECTON_RELEASES_URL, TECTON_LATEST_RELEASE_URL
[docs]
class TectonResponse:
[docs]
def to_json(self):
return vars(self)
[docs]
class Success(TectonResponse):
"""
API result that represents a Okay (200) response.
"""
def __init__(self, data: Optional[dict] = None, status: Optional[int] = 200):
self.statusText = "success"
self.data = {} if data is None else data
self.status = status
[docs]
class InternalServerError(TectonResponse):
"""
API result that represents a Internal Server Error (500) response.
"""
def __init__(
self, message: str, data: Optional[dict] = None, status: Optional[int] = 500
):
self.statusText = "fail"
self.message = message
self.data = {} if data is None else data
self.status = status
[docs]
class BadRequest(TectonResponse):
"""
API result that represents a Bad Request (400) response.
"""
def __init__(
self, message: str, data: Optional[dict] = None, status: Optional[int] = 400
):
self.statusText = "error"
self.message = message
self.data = {} if data is None else data
self.status = status
[docs]
class NotFound(TectonResponse):
"""
API result that represents a Not Found (404) response.
"""
def __init__(self, data: Optional[dict] = None, status: Optional[int] = 404):
self.statusText = "error"
self.data = {} if data is None else data
self.status = status
[docs]
@dataclass
class TectonVersion:
patch_version: str
def __init__(self, versions_dict: dict):
self.js_path = versions_dict.get("sdkJS")
self.canary_url = versions_dict.get("canaryURL")
self.docs_url = versions_dict.get("docsURL")
self.patch_version = versions_dict.get("patchVersion")
self.shell_url = versions_dict.get("shellURL")
self.version = versions_dict.get("version")
[docs]
@staticmethod
async def get_all_versions(logger: Optional[Q2LoggerType] = None):
all_versions = await q2_requests.get(
logger, TECTON_RELEASES_URL, verify_whitelist=False
)
return [TectonVersion(x) for x in all_versions.json() if "sdkJS" in x]
[docs]
@staticmethod
async def get_latest_version(logger: Optional[Q2LoggerType] = None):
current_version = await q2_requests.get(
logger, TECTON_LATEST_RELEASE_URL, verify_whitelist=False
)
return TectonVersion(current_version.json())