(
self,
*,
job_queue: str = "codeclash-queue",
region: str = "us-east-1",
logs_base_dir: Path | None = None,
)
| 23 | """Monitor AWS Batch jobs""" |
| 24 | |
| 25 | def __init__( |
| 26 | self, |
| 27 | *, |
| 28 | job_queue: str = "codeclash-queue", |
| 29 | region: str = "us-east-1", |
| 30 | logs_base_dir: Path | None = None, |
| 31 | ): |
| 32 | if boto3 is None: |
| 33 | msg = "boto3 is not installed. Install it with: pip install codeclash[aws]" |
| 34 | raise ImportError(msg) |
| 35 | self.batch_client = boto3.client("batch", region_name=region) |
| 36 | self.job_queue = job_queue |
| 37 | self.region = region |
| 38 | self.logs_base_dir = logs_base_dir or Path("logs") |
| 39 | self._job_id_to_folder: dict[str, str] | None = None |
| 40 | self._job_id_to_round_info: dict[str, tuple[int, int] | None] | None = None |
| 41 | self._job_id_to_aws_command: dict[str, str | None] | None = None |
| 42 | |
| 43 | def list_jobs(self, *, limit: int | None = None, hours_back: int = 24) -> list[dict[str, Any]]: |
| 44 | """List all jobs from AWS Batch |
nothing calls this directly
no outgoing calls
no test coverage detected