I am trying to submit batch job and send to it parameters by boto3. Unfortunately the parameters don't pass.
Here is what I am trying to do:
class AWSBatch:
def __init__(self, job_queue: str, job_definition: str) -> None:
self.job_queue = job_queue
self.job_definition = job_definition
self.client = boto3.client("batch")
def run(self, job_name: str, dependencies: List[Dict[str, str]] = None, parameters: Dict[str, str] = None,
**kwargs
) -> Dict[str, Any]:
try:
submit_job_params: Dict[str, Any] = {
"jobName": job_name,
"jobQueue": self.job_queue,
"jobDefinition": self.job_definition
}
if dependencies:
submit_job_params['dependsOn'] = dependencies
if parameters:
submit_job_params['parameters'] = parameters
submit_job_params.update(kwargs)
response: Dict[str, Any] = self.client.submit_job(**submit_job_params)
response = {
"status": True,
"data": response,
}
except Exception as e:
response = {"status": False, "message": str(e)}
return response