How to pass arguments containing parentheses to aws ecs execute-command using python subprocess.run?

Question

Goal: Use the aws cli in a subprocess.run call to execute a command in a container of a running task.
Example Usage:

task_arn = 'a_task_arn'
arg = event = {'something': 'that contains ( )'}

# Attempt 1: json string
arg = json.dumps(arg)
subprocess_cmd = 'aws ecs execute-command --command "python -m task.function {arg}" --interactive --task {task_arn} --container Container'
subprocess.run(subprocess_cmd, shell=True

# Attempt 2: encoded json string
arg = json.dumps(arg).encode('utf-8')
subprocess_cmd = 'aws ecs execute-command --command "python -m task.function {arg}" --interactive --task {task_arn} --container Container'
subprocess.run(subprocess_cmd, shell=True

Encountered Error: /bin/sh: 1: Syntax error: "(" unexpected

I am trying to use the cli through a subprocess.run call instead of using the boto3 execute_command because

• it doesn't appear that there is an easy way to read the websocket boto3 returns: https://github.com/boto/boto3/issues/3496,
• as indicated in this SO answer, it only returns a very small portion of the websocket message: How can I get output from boto3 ecs execute_command?

Answer 1

The big advantage that subprocess has over os.system is that it lets you pass an explicit argument vector, so you don't need to worry about how a shell is going to parse your command. Where you can't take advantage of that, Python provides shlex.quote() to convert a string into code that any POSIX-compliant shell will evaluate into that same string.

import json, shlex, subprocess

task_arn = 'a_task_arn'

arg = {'something': 'that contains ( )'}
arg_json = json.dumps(arg)

# to locally run the command we could subprocess.run(remote_argv)
remote_argv = [ 'python', '-m', 'task.function', arg_json ]

remote_str = ' '.join([shlex.quote(s) for s in remote_argv])

local_argv = [
  'aws', 'ecs', 'execute-command',
  '--command', remote_str,
  '--interactive',
  '--task', task_arn,
  '--container', 'Container'
]
subprocess.run(local_argv) # WITHOUT shell=True