Restore a specific version of an environment from history Args: env_name: Name of the environment version: Version string to restore auto_initialize: Whether to automatically initialize the restored environment Returns:
(self, env_name: str, version: str, auto_initialize: bool = True)
| 1124 | |
| 1125 | |
| 1126 | async def restore(self, env_name: str, version: str, auto_initialize: bool = True) -> Optional[EnvironmentConfig]: |
| 1127 | """Restore a specific version of an environment from history |
| 1128 | |
| 1129 | Args: |
| 1130 | env_name: Name of the environment |
| 1131 | version: Version string to restore |
| 1132 | auto_initialize: Whether to automatically initialize the restored environment |
| 1133 | |
| 1134 | Returns: |
| 1135 | EnvironmentConfig of the restored version, or None if not found |
| 1136 | """ |
| 1137 | # Look up version from dict-based history (O(1) lookup) |
| 1138 | version_config = None |
| 1139 | if env_name in self._environment_history_versions: |
| 1140 | version_config = self._environment_history_versions[env_name].get(version) |
| 1141 | |
| 1142 | if version_config is None: |
| 1143 | logger.warning(f"| ⚠️ Version {version} not found for environment {env_name}") |
| 1144 | return None |
| 1145 | |
| 1146 | # Create a copy to avoid modifying the history |
| 1147 | restored_config = EnvironmentConfig(**version_config.model_dump()) |
| 1148 | |
| 1149 | # Set as current active config |
| 1150 | self._environment_configs[env_name] = restored_config |
| 1151 | |
| 1152 | # Update version manager current version |
| 1153 | version_history = await version_manager.get_version_history("environment", env_name) |
| 1154 | if version_history: |
| 1155 | # Check if version exists in version history, if not register it |
| 1156 | if version not in version_history.versions: |
| 1157 | await version_manager.register_version("environment", env_name, version) |
| 1158 | version_history.current_version = version |
| 1159 | else: |
| 1160 | # If version history doesn't exist, register the version first |
| 1161 | await version_manager.register_version("environment", env_name, version) |
| 1162 | |
| 1163 | # Initialize if requested |
| 1164 | if auto_initialize and restored_config.cls is not None: |
| 1165 | await self.build(restored_config) |
| 1166 | |
| 1167 | # Persist to JSON (current_version changes) |
| 1168 | await self.save_to_json() |
| 1169 | |
| 1170 | logger.info(f"| 🔄 Restored environment {env_name} to version {version}") |
| 1171 | return restored_config |
| 1172 | |
| 1173 | async def save_contract(self, env_names: Optional[List[str]] = None): |
| 1174 | """Save the contract for an environment""" |
nothing calls this directly
no test coverage detected