| 111 | |
| 112 | |
| 113 | class ConnectionsStore: |
| 114 | def __init__(self, storage: ConnectionsStorage | None = None) -> None: |
| 115 | self._storage = storage or FileConnectionsStorage() |
| 116 | self._primary = _sanitize_primary(self._storage.load()) |
| 117 | |
| 118 | def primary(self) -> PrimaryConnection: |
| 119 | return self._primary |
| 120 | |
| 121 | def export(self) -> dict[str, Any]: |
| 122 | return { |
| 123 | "source": "local-file", |
| 124 | "editable": True, |
| 125 | "primary": { |
| 126 | "upstream": self._primary.upstream, |
| 127 | "has_api_key": bool(self._primary.api_key), |
| 128 | }, |
| 129 | } |
| 130 | |
| 131 | def set_primary( |
| 132 | self, |
| 133 | *, |
| 134 | upstream: str, |
| 135 | api_key: str, |
| 136 | ) -> dict[str, Any]: |
| 137 | self._primary = PrimaryConnection( |
| 138 | upstream=_clean_text(upstream), |
| 139 | api_key=_clean_text(api_key), |
| 140 | ) |
| 141 | self._persist() |
| 142 | return self.export() |
| 143 | |
| 144 | def reset(self) -> dict[str, Any]: |
| 145 | self._primary = PrimaryConnection() |
| 146 | self._persist() |
| 147 | return self.export() |
| 148 | |
| 149 | def _persist(self) -> None: |
| 150 | self._storage.save({ |
| 151 | "primary": { |
| 152 | "upstream": self._primary.upstream, |
| 153 | "api_key": self._primary.api_key, |
| 154 | }, |
| 155 | }) |
| 156 | |
| 157 | |
| 158 | def resolve_primary_connection( |
no outgoing calls