Changeset View
Changeset View
Standalone View
Standalone View
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): | ||||
| pass | def __init__(self, *args, body=None): | ||||
| super().__init__(*args) | |||||
| self.body = body | |||||
| class ReloadRepositoryLocationStatus(Enum): | class ReloadRepositoryLocationStatus(Enum): | ||||
| """This enum describes the status of a GraphQL mutation to reload a Dagster repository location | """This enum describes the status of a GraphQL mutation to reload a Dagster repository location | ||||
| Args: | Args: | ||||
| Enum (str): can be either `ReloadRepositoryLocationStatus.SUCCESS` | Enum (str): can be either `ReloadRepositoryLocationStatus.SUCCESS` | ||||
| or `ReloadRepositoryLocationStatus.FAILURE`. | or `ReloadRepositoryLocationStatus.FAILURE`. | ||||
| Show All 10 Lines | class ReloadRepositoryLocationInfo(NamedTuple): | ||||
| Args: | Args: | ||||
| status (ReloadRepositoryLocationStatus): The status of the reload repository location mutation | status (ReloadRepositoryLocationStatus): The status of the reload repository location mutation | ||||
| message (Optional[str], optional): the failure message/reason if | message (Optional[str], optional): the failure message/reason if | ||||
| `status == ReloadRepositoryLocationStatus.FAILURE`. Defaults to None. | `status == ReloadRepositoryLocationStatus.FAILURE`. Defaults to None. | ||||
| """ | """ | ||||
| status: ReloadRepositoryLocationStatus | status: ReloadRepositoryLocationStatus | ||||
| message: Optional[str] = None | message: Optional[str] = None | ||||
| class PipelineInfo(NamedTuple): | |||||
rexledesma: this is more about the information for a specific a pipeline, rather than the repository or… | |||||
| repository_location_name: str | |||||
| repository_name: str | |||||
| pipeline_name: str | |||||
| @staticmethod | |||||
| def from_node(node: Dict[str, Any]) -> List["PipelineInfo"]: | |||||
| repo_name = node["name"] | |||||
| repo_location_name = node["location"]["name"] | |||||
| return [ | |||||
| PipelineInfo( | |||||
| repository_location_name=repo_location_name, | |||||
| repository_name=repo_name, | |||||
| pipeline_name=pipeline["name"], | |||||
| ) | |||||
| for pipeline in node["pipelines"] | |||||
| ] | |||||
| class InvalidOutputErrorInfo(NamedTuple): | |||||
| """This class gives information about an InvalidOutputError from submitting a pipeline for execution | |||||
| from GraphQL. | |||||
| Args: | |||||
| step_key (str): key of the step that failed | |||||
| invalid_output_name (str): the name of the invalid output from the given step | |||||
| """ | |||||
| step_key: str | |||||
| invalid_output_name: str | |||||
this is more about the information for a specific a pipeline, rather than the repository or repository location