Changeset View
Changeset View
Standalone View
Standalone View
python_modules/dagster/dagster/core/host_representation/selector.py
| Show First 20 Lines • Show All 84 Lines • ▼ Show 20 Lines | ): | ||||
| def from_graphql_input(graphql_data): | def from_graphql_input(graphql_data): | ||||
| return ScheduleSelector( | return ScheduleSelector( | ||||
| location_name=graphql_data["repositoryLocationName"], | location_name=graphql_data["repositoryLocationName"], | ||||
| repository_name=graphql_data["repositoryName"], | repository_name=graphql_data["repositoryName"], | ||||
| schedule_name=graphql_data["scheduleName"], | schedule_name=graphql_data["scheduleName"], | ||||
| ) | ) | ||||
| class TriggerSelector( | class TriggerSelector(namedtuple("_TriggerSelector", "location_name repository_name job_name")): | ||||
| namedtuple("_TriggerSelector", "location_name repository_name executable_name") | def __new__(cls, location_name, repository_name, job_name): | ||||
| ): | |||||
| def __new__(cls, location_name, repository_name, executable_name): | |||||
| return super(TriggerSelector, cls).__new__( | return super(TriggerSelector, cls).__new__( | ||||
| cls, | cls, | ||||
| location_name=check.str_param(location_name, "location_name"), | location_name=check.str_param(location_name, "location_name"), | ||||
| repository_name=check.str_param(repository_name, "repository_name"), | repository_name=check.str_param(repository_name, "repository_name"), | ||||
| executable_name=check.str_param(executable_name, "executable_name"), | job_name=check.str_param(job_name, "job_name"), | ||||
| ) | ) | ||||
| def to_graphql_input(self): | def to_graphql_input(self): | ||||
| return { | return { | ||||
| "repositoryLocationName": self.location_name, | "repositoryLocationName": self.location_name, | ||||
| "repositoryName": self.repository_name, | "repositoryName": self.repository_name, | ||||
| "executableName": self.executable_name, | "jobName": self.job_name, | ||||
| } | } | ||||
| @staticmethod | @staticmethod | ||||
| def from_graphql_input(graphql_data): | def from_graphql_input(graphql_data): | ||||
| return TriggerSelector( | return TriggerSelector( | ||||
| location_name=graphql_data["repositoryLocationName"], | location_name=graphql_data["repositoryLocationName"], | ||||
| repository_name=graphql_data["repositoryName"], | repository_name=graphql_data["repositoryName"], | ||||
| executable_name=graphql_data["executableName"], | job_name=graphql_data["jobName"], | ||||
| ) | ) | ||||