| 99 | |
| 100 | |
| 101 | class GSConfig(object): |
| 102 | def __init__(self, contexts, current_context: str): |
| 103 | self._contexts = contexts |
| 104 | self._current_context = current_context |
| 105 | |
| 106 | def current_context(self) -> Context: |
| 107 | if self._current_context is None: |
| 108 | return None |
| 109 | if self._current_context not in self._contexts: |
| 110 | raise RuntimeError( |
| 111 | f"Failed to get current context: {self._current_context}" |
| 112 | ) |
| 113 | return self._contexts[self._current_context] |
| 114 | |
| 115 | def set_and_write(self, context: Context): |
| 116 | # treat the same endpoint with same services as the same coordinator |
| 117 | for _, v in self._contexts.items(): |
| 118 | if ( |
| 119 | context.coordinator_endpoint == v.coordinator_endpoint |
| 120 | and context.flex == v.flex |
| 121 | ): |
| 122 | # reset to global context |
| 123 | v.switch_context("global") |
| 124 | contexts = [v.to_dict() for _, v in self._contexts.items()] |
| 125 | write_yaml_file( |
| 126 | {"contexts": contexts, "current-context": self._current_context}, |
| 127 | GS_CONFIG_DEFAULT_LOCATION, |
| 128 | ) |
| 129 | return |
| 130 | |
| 131 | # set |
| 132 | self._current_context = context.name |
| 133 | self._contexts[context.name] = context |
| 134 | |
| 135 | # write |
| 136 | contexts = [v.to_dict() for _, v in self._contexts.items()] |
| 137 | write_yaml_file( |
| 138 | {"contexts": contexts, "current-context": self._current_context}, |
| 139 | GS_CONFIG_DEFAULT_LOCATION, |
| 140 | ) |
| 141 | |
| 142 | def update_and_write(self, context: Context): |
| 143 | if context.name not in self._contexts: |
| 144 | raise RuntimeError(f"Failed to get context: {context.name}") |
| 145 | |
| 146 | # update |
| 147 | self._contexts[context.name] = context |
| 148 | |
| 149 | # write |
| 150 | contexts = [v.to_dict() for _, v in self._contexts.items()] |
| 151 | write_yaml_file( |
| 152 | {"contexts": contexts, "current-context": self._current_context}, |
| 153 | GS_CONFIG_DEFAULT_LOCATION, |
| 154 | ) |
| 155 | |
| 156 | def remove_and_write(self, current_context: Context): |
| 157 | # remove |
| 158 | del self._contexts[current_context.name] |