Page MenuHomeElementl
Differential D7424 Diff 35380 python_modules/dagster-graphql/dagster_graphql/client/utils.py

Changeset View

Standalone View

View Options

python_modules/dagster-graphql/dagster_graphql/client/utils.py

from enum import Enum from enum import Enum
from typing import NamedTuple, Optional from typing import Any, Dict, List, NamedTuple, Optional
class DagsterGraphQLClientError(Exception): class DagsterGraphQLClientError(Exception):
def __init__(self, *args, body=None): def __init__(self, *args, body=None):
super(DagsterGraphQLClientError, self).__init__(*args) super(DagsterGraphQLClientError, self).__init__(*args)
self.body = body self.body = body
class ReloadRepositoryLocationStatus(Enum): class ReloadRepositoryLocationStatus(Enum):
SUCCESS = "SUCCESS" SUCCESS = "SUCCESS"
FAILURE = "FAILURE" FAILURE = "FAILURE"
class ReloadRepositoryLocationInfo(NamedTuple): class ReloadRepositoryLocationInfo(NamedTuple):
status: ReloadRepositoryLocationStatus status: ReloadRepositoryLocationStatus
message: Optional[str] = None message: Optional[str] = None
class RepositoryNode(NamedTuple):
rexledesmaUnsubmitted
Done
Inline Actions
message: Optional[str] = None
- class RepositoryNodeTuple(NamedTuple):
+ class PipelineInfo(NamedTuple):
repository_location_name: str

this is more about the information for a specific a pipeline, rather than the repository or repository location

rexledesma: this is more about the information for a specific a pipeline, rather than the repository or…
repository_location_name: str
repository_names: List[str]
@staticmethod
def from_dict(obj: Dict[str, Any]):
repository_names: List[str] = [repository["name"] for repository in obj["repositories"]]
return RepositoryNode(
repository_location_name=obj["name"], repository_names=repository_names
)