Coverage for src/container_collection/batch/get_batch_logs.py: 100%

1import boto3 

2 

3LOG_GROUP_NAME = "/aws/batch/job" 

4"""AWS Batch log group name.""" 

5 

6 

7def get_batch_logs(job_arn: str, log_filter: str) -> str: 

8 """ 

9 Get logs for AWS Batch job. 

10 

11 Parameters 

12 ---------- 

13 job_arn 

14 Job ARN. 

15 log_filter 

16 Filter for log events. 

17 

18 Returns 

19 ------- 

20 : 

21 All filtered log events. 

22 """ 

23 

24 client = boto3.client("batch") 

25 response = client.describe_jobs(jobs=[job_arn])["jobs"][0] 

26 log_stream = response["container"]["logStreamName"] 

27 

28 client = boto3.client("logs") 

29 log_events: list[str] = [] 

30 

31 response = client.filter_log_events( 

32 logGroupName=LOG_GROUP_NAME, 

33 logStreamNames=[log_stream], 

34 filterPattern=log_filter, 

35 ) 

36 

37 if response["events"]: 

38 log_events = log_events + [event["message"] for event in response["events"]] 

39 

40 while "nextToken" in response: 

41 response = client.filter_log_events( 

42 logGroupName="/aws/batch/job", 

43 logStreamNames=[log_stream], 

44 filterPattern=log_filter, 

45 nextToken=response["nextToken"], 

46 ) 

47 

48 log_events = log_events + [event["message"] for event in response["events"]] 

49 

50 return "\n".join(log_events)