Coverage for src/container_collection/fargate/submit_fargate_task.py: 100%

1from __future__ import annotations 

2 

3import boto3 

4 

5 

6def submit_fargate_task( 

7 name: str, 

8 task_definition_arn: str, 

9 user: str, 

10 cluster: str, 

11 security_groups: list[str], 

12 subnets: list[str], 

13 command: list[str], 

14 **kwargs: list | str | int | bool | dict, 

15) -> list[str]: 

16 """ 

17 Submit task to AWS Fargate. 

18 

19 Parameters 

20 ---------- 

21 name 

22 Task name. 

23 task_definition_arn 

24 Task definition ARN. 

25 user 

26 User name prefix for task name. 

27 cluster 

28 ECS cluster name. 

29 security_groups 

30 List of security groups. 

31 subnets 

32 List of subnets. 

33 command 

34 Command list passed to container. 

35 **kwargs 

36 Additional parameters for task submission. The keyword arguments are 

37 passed to `boto3` ECS client method `run_task`. 

38 

39 Returns 

40 ------- 

41 : 

42 Task ARN. 

43 """ 

44 

45 default_task_submission = { 

46 "taskDefinition": task_definition_arn, 

47 "capacityProviderStrategy": [ 

48 {"capacityProvider": "FARGATE", "weight": 1}, 

49 {"capacityProvider": "FARGATE_SPOT", "weight": 1}, 

50 ], 

51 "cluster": cluster, 

52 "platformVersion": "LATEST", 

53 "count": 1, 

54 "networkConfiguration": { 

55 "awsvpcConfiguration": { 

56 "subnets": subnets, 

57 "assignPublicIp": "ENABLED", 

58 "securityGroups": security_groups, 

59 } 

60 }, 

61 "overrides": { 

62 "containerOverrides": [ 

63 { 

64 "name": f"{user}_{name}", 

65 "command": command, 

66 } 

67 ] 

68 }, 

69 } 

70 

71 client = boto3.client("ecs") 

72 task_submission = default_task_submission | kwargs 

73 response = client.run_task(**task_submission) 

74 

75 return response["tasks"][0]["taskArn"]