Changeset View
Changeset View
Standalone View
Standalone View
python_modules/dagster/dagster/cli/workspace/context.py
| Show First 20 Lines • Show All 64 Lines • ▼ Show 20 Lines | class WorkspaceRequestContext(NamedTuple): | ||||
| @property | @property | ||||
| def read_only(self) -> bool: | def read_only(self) -> bool: | ||||
| return self.process_context.read_only | return self.process_context.read_only | ||||
| def repository_location_errors(self) -> List["SerializableErrorInfo"]: | def repository_location_errors(self) -> List["SerializableErrorInfo"]: | ||||
| return [entry.load_error for entry in self.workspace_snapshot.values() if entry.load_error] | return [entry.load_error for entry in self.workspace_snapshot.values() if entry.load_error] | ||||
| def get_repository_location(self, name: str) -> RepositoryLocation: | def get_repository_location(self, name: str) -> RepositoryLocation: | ||||
| return self.workspace_snapshot[name].repository_location | repository_location = self.workspace_snapshot[name].repository_location | ||||
| assert isinstance( | |||||
| repository_location, RepositoryLocation | |||||
| ), f"Couldn't get repository location for {name}" | |||||
| return repository_location | |||||
| def get_load_status(self, name: str) -> WorkspaceLocationLoadStatus: | def get_load_status(self, name: str) -> WorkspaceLocationLoadStatus: | ||||
| return self.workspace_snapshot[name].load_status | return self.workspace_snapshot[name].load_status | ||||
| def has_repository_location_error(self, name: str) -> bool: | def has_repository_location_error(self, name: str) -> bool: | ||||
| return self.get_repository_location_error(name) != None | return self.get_repository_location_error(name) != None | ||||
| def get_repository_location_error(self, name: str) -> "SerializableErrorInfo": | def get_repository_location_error(self, name: str) -> "SerializableErrorInfo": | ||||
| return self.workspace_snapshot[name].load_error | load_error = self.workspace_snapshot[name].load_error | ||||
| assert isinstance( | |||||
| load_error, SerializableErrorInfo | |||||
Lint: Used Before Assignment: Using variable 'SerializableErrorInfo' before assignment | |||||
| ), f"No load error for workspace snapshot {name}" | |||||
| return load_error | |||||
| def has_repository_location(self, name: str) -> bool: | def has_repository_location(self, name: str) -> bool: | ||||
| return self.get_repository_location(name) != None | return self.get_repository_location(name) != None | ||||
| def is_reload_supported(self, name: str) -> bool: | def is_reload_supported(self, name: str) -> bool: | ||||
| return self.workspace_snapshot[name].origin.is_reload_supported | return self.workspace_snapshot[name].origin.is_reload_supported | ||||
| def reload_repository_location(self, name: str) -> "WorkspaceRequestContext": | def reload_repository_location(self, name: str) -> "WorkspaceRequestContext": | ||||
| ▲ Show 20 Lines • Show All 173 Lines • Show Last 20 Lines | |||||
Using variable 'SerializableErrorInfo' before assignment