Source code for q2_sdk.models.tecton

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())