Get the state of an environment Args: env_name: Environment name ctx: Environment context Returns: Optional[Dict[str, Any]]: State of the environment or None if not found
(self, env_name: str, ctx: SessionContext = None, **kwargs)
| 614 | return self._environment_configs.get(env_name) |
| 615 | |
| 616 | async def get_state(self, env_name: str, ctx: SessionContext = None, **kwargs) -> Optional[Dict[str, Any]]: |
| 617 | """Get the state of an environment |
| 618 | |
| 619 | Args: |
| 620 | env_name: Environment name |
| 621 | ctx: Environment context |
| 622 | Returns: |
| 623 | Optional[Dict[str, Any]]: State of the environment or None if not found |
| 624 | """ |
| 625 | |
| 626 | if ctx is None: |
| 627 | ctx = SessionContext() |
| 628 | |
| 629 | env_args = { |
| 630 | "ctx": ctx, |
| 631 | } |
| 632 | |
| 633 | env_config = self._environment_configs.get(env_name) |
| 634 | if not env_config or not env_config.instance: |
| 635 | raise ValueError(f"Environment '{env_name}' not found") |
| 636 | return await env_config.instance.get_state(**env_args) |
| 637 | |
| 638 | async def list(self) -> List[str]: |
| 639 | """Get list of registered environments |
nothing calls this directly
no test coverage detected