We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
RDS call modify_dbinstance with multi_az = True doesn't actually set an instance to MultiAZ
Making a call to a non-multiaz instance with multi_az=True doesn't actually switch the parameter. I assume this is also true for creating one from scratch, but I haven't tested that yet.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `boto/emr/connection.py`
Content:
```
1 # Copyright (c) 2010 Spotify AB
2 #
3 # Permission is hereby granted, free of charge, to any person obtaining a
4 # copy of this software and associated documentation files (the
5 # "Software"), to deal in the Software without restriction, including
6 # without limitation the rights to use, copy, modify, merge, publish, dis-
7 # tribute, sublicense, and/or sell copies of the Software, and to permit
8 # persons to whom the Software is furnished to do so, subject to the fol-
9 # lowing conditions:
10 #
11 # The above copyright notice and this permission notice shall be included
12 # in all copies or substantial portions of the Software.
13 #
14 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
15 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
16 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
17 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
18 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
20 # IN THE SOFTWARE.
21 
22 """
23 Represents a connection to the EMR service
24 """
25 import types
26 
27 import boto
28 from boto.ec2.regioninfo import RegionInfo
29 from boto.emr.emrobject import JobFlow, RunJobFlowResponse
30 from boto.emr.step import JarStep
31 from boto.connection import AWSQueryConnection
32 from boto.exception import EmrResponseError
33 
34 class EmrConnection(AWSQueryConnection):
35 
36     APIVersion = boto.config.get('Boto', 'emr_version', '2009-03-31')
37     DefaultRegionName = boto.config.get('Boto', 'emr_region_name', 'us-east-1')
38     DefaultRegionEndpoint = boto.config.get('Boto', 'emr_region_endpoint',
39                                             'elasticmapreduce.amazonaws.com')
40     ResponseError = EmrResponseError
41 
42     # Constants for AWS Console debugging
43     DebuggingJar = 's3n://us-east-1.elasticmapreduce/libs/script-runner/script-runner.jar'
44     DebuggingArgs = 's3n://us-east-1.elasticmapreduce/libs/state-pusher/0.1/fetch'
45 
46     def __init__(self, aws_access_key_id=None, aws_secret_access_key=None,
47                  is_secure=True, port=None, proxy=None, proxy_port=None,
48                  proxy_user=None, proxy_pass=None, debug=0,
49                  https_connection_factory=None, region=None, path='/'):
50         if not region:
51             region = RegionInfo(self, self.DefaultRegionName, self.DefaultRegionEndpoint)
52         self.region = region
53         AWSQueryConnection.__init__(self, aws_access_key_id,
54                                     aws_secret_access_key,
55                                     is_secure, port, proxy, proxy_port,
56                                     proxy_user, proxy_pass,
57                                     self.region.endpoint, debug,
58                                     https_connection_factory, path)
59 
60     def _required_auth_capability(self):
61         return ['emr']
62 
63     def describe_jobflow(self, jobflow_id):
64         """
65         Describes a single Elastic MapReduce job flow
66 
67         :type jobflow_id: str
68         :param jobflow_id: The job flow id of interest
69         """
70         jobflows = self.describe_jobflows(jobflow_ids=[jobflow_id])
71         if jobflows:
72             return jobflows[0]
73 
74     def describe_jobflows(self, states=None, jobflow_ids=None,
75                            created_after=None, created_before=None):
76         """
77         Retrieve all the Elastic MapReduce job flows on your account
78 
79         :type states: list
80         :param states: A list of strings with job flow states wanted
81 
82         :type jobflow_ids: list
83         :param jobflow_ids: A list of job flow IDs
84         :type created_after: datetime
85         :param created_after: Bound on job flow creation time
86 
87         :type created_before: datetime
88         :param created_before: Bound on job flow creation time
89         """
90         params = {}
91 
92         if states:
93             self.build_list_params(params, states, 'JobFlowStates.member')
94         if jobflow_ids:
95             self.build_list_params(params, jobflow_ids, 'JobFlowIds.member')
96         if created_after:
97             params['CreatedAfter'] = created_after.strftime('%Y-%m-%dT%H:%M:%S')
98         if created_before:
99             params['CreatedBefore'] = created_before.strftime('%Y-%m-%dT%H:%M:%S')
100 
101         return self.get_list('DescribeJobFlows', params, [('member', JobFlow)])
102 
103     def terminate_jobflow(self, jobflow_id):
104         """
105         Terminate an Elastic MapReduce job flow
106 
107         :type jobflow_id: str
108         :param jobflow_id: A jobflow id 
109         """
110         self.terminate_jobflows([jobflow_id]) 
111 
112     def terminate_jobflows(self, jobflow_ids):
113         """
114         Terminate an Elastic MapReduce job flow
115 
116         :type jobflow_ids: list
117         :param jobflow_ids: A list of job flow IDs
118         """
119         params = {}
120         self.build_list_params(params, jobflow_ids, 'JobFlowIds.member')
121         return self.get_status('TerminateJobFlows', params)
122 
123     def add_jobflow_steps(self, jobflow_id, steps):
124         """
125         Adds steps to a jobflow
126 
127         :type jobflow_id: str
128         :param jobflow_id: The job flow id
129         :type steps: list(boto.emr.Step)
130         :param steps: A list of steps to add to the job
131         """
132         if type(steps) != types.ListType:
133             steps = [steps]
134         params = {}
135         params['JobFlowId'] = jobflow_id
136 
137         # Step args
138         step_args = [self._build_step_args(step) for step in steps]
139         params.update(self._build_step_list(step_args))
140 
141         return self.get_object('AddJobFlowSteps', params, RunJobFlowResponse)
142 
143     def run_jobflow(self, name, log_uri, ec2_keyname=None, availability_zone=None,
144                     master_instance_type='m1.small',
145                     slave_instance_type='m1.small', num_instances=1,
146                     action_on_failure='TERMINATE_JOB_FLOW', keep_alive=False,
147                     enable_debugging=False,
148                     hadoop_version='0.18',
149                     steps=[],
150                     bootstrap_actions=[]):
151         """
152         Runs a job flow
153 
154         :type name: str
155         :param name: Name of the job flow
156         :type log_uri: str
157         :param log_uri: URI of the S3 bucket to place logs
158         :type ec2_keyname: str
159         :param ec2_keyname: EC2 key used for the instances
160         :type availability_zone: str
161         :param availability_zone: EC2 availability zone of the cluster
162         :type master_instance_type: str
163         :param master_instance_type: EC2 instance type of the master
164         :type slave_instance_type: str
165         :param slave_instance_type: EC2 instance type of the slave nodes
166         :type num_instances: int
167         :param num_instances: Number of instances in the Hadoop cluster
168         :type action_on_failure: str
169         :param action_on_failure: Action to take if a step terminates
170         :type keep_alive: bool
171         :param keep_alive: Denotes whether the cluster should stay alive upon completion
172         :type enable_debugging: bool
173         :param enable_debugging: Denotes whether AWS console debugging should be enabled.
174         :type steps: list(boto.emr.Step)
175         :param steps: List of steps to add with the job
176 
177         :rtype: str
178         :return: The jobflow id
179         """
180         params = {}
181         if action_on_failure:
182             params['ActionOnFailure'] = action_on_failure
183         params['Name'] = name
184         params['LogUri'] = log_uri
185 
186         # Instance args
187         instance_params = self._build_instance_args(ec2_keyname, availability_zone,
188                                                     master_instance_type, slave_instance_type,
189                                                     num_instances, keep_alive, hadoop_version)
190         params.update(instance_params)
191 
192         # Debugging step from EMR API docs
193         if enable_debugging:
194             debugging_step = JarStep(name='Setup Hadoop Debugging',
195                                      action_on_failure='TERMINATE_JOB_FLOW',
196                                      main_class=None,
197                                      jar=self.DebuggingJar,
198                                      step_args=self.DebuggingArgs)
199             steps.insert(0, debugging_step)
200 
201         # Step args
202         if steps:
203             step_args = [self._build_step_args(step) for step in steps]
204             params.update(self._build_step_list(step_args))
205 
206         if bootstrap_actions:
207             bootstrap_action_args = [self._build_bootstrap_action_args(bootstrap_action) for bootstrap_action in bootstrap_actions]
208             params.update(self._build_bootstrap_action_list(bootstrap_action_args))
209 
210         response = self.get_object('RunJobFlow', params, RunJobFlowResponse)
211         return response.jobflowid
212 
213     def _build_bootstrap_action_args(self, bootstrap_action):
214         bootstrap_action_params = {}
215         bootstrap_action_params['ScriptBootstrapAction.Path'] = bootstrap_action.path
216 
217         try:
218             bootstrap_action_params['Name'] = bootstrap_action.name
219         except AttributeError:
220             pass
221 
222         args = bootstrap_action.args()
223         if args:
224             self.build_list_params(bootstrap_action_params, args, 'ScriptBootstrapAction.Args.member')
225 
226         return bootstrap_action_params
227 
228     def _build_step_args(self, step):
229         step_params = {}
230         step_params['ActionOnFailure'] = step.action_on_failure
231         step_params['HadoopJarStep.Jar'] = step.jar()
232 
233         main_class = step.main_class()
234         if main_class:
235             step_params['HadoopJarStep.MainClass'] = main_class
236 
237         args = step.args()
238         if args:
239             self.build_list_params(step_params, args, 'HadoopJarStep.Args.member')
240 
241         step_params['Name'] = step.name
242         return step_params
243 
244     def _build_bootstrap_action_list(self, bootstrap_actions):
245         if type(bootstrap_actions) != types.ListType:
246             bootstrap_actions = [bootstrap_actions]
247 
248         params = {}
249         for i, bootstrap_action in enumerate(bootstrap_actions):
250             for key, value in bootstrap_action.iteritems():
251                 params['BootstrapActions.member.%s.%s' % (i + 1, key)] = value
252         return params
253 
254     def _build_step_list(self, steps):
255         if type(steps) != types.ListType:
256             steps = [steps]
257 
258         params = {}
259         for i, step in enumerate(steps):
260             for key, value in step.iteritems():
261                 params['Steps.member.%s.%s' % (i+1, key)] = value
262         return params
263 
264     def _build_instance_args(self, ec2_keyname, availability_zone, master_instance_type,
265                              slave_instance_type, num_instances, keep_alive, hadoop_version):
266         params = {
267             'Instances.MasterInstanceType' : master_instance_type,
268             'Instances.SlaveInstanceType' : slave_instance_type,
269             'Instances.InstanceCount' : num_instances,
270             'Instances.KeepJobFlowAliveWhenNoSteps' : str(keep_alive).lower(),
271             'Instances.HadoopVersion' : hadoop_version
272         }
273 
274         if ec2_keyname:
275             params['Instances.Ec2KeyName'] = ec2_keyname
276         if availability_zone:
277             params['Placement'] = availability_zone
278 
279         return params
280 
281
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/boto/emr/connection.py b/boto/emr/connection.py
--- a/boto/emr/connection.py
+++ b/boto/emr/connection.py
@@ -274,7 +274,7 @@
         if ec2_keyname:
             params['Instances.Ec2KeyName'] = ec2_keyname
         if availability_zone:
-            params['Placement'] = availability_zone
+            params['Placement.AvailabilityZone'] = availability_zone
 
         return params

{"golden_diff": "diff --git a/boto/emr/connection.py b/boto/emr/connection.py\n--- a/boto/emr/connection.py\n+++ b/boto/emr/connection.py\n@@ -274,7 +274,7 @@\n         if ec2_keyname:\n             params['Instances.Ec2KeyName'] = ec2_keyname\n         if availability_zone:\n-            params['Placement'] = availability_zone\n+            params['Placement.AvailabilityZone'] = availability_zone\n \n         return params\n", "issue": "RDS call modify_dbinstance with multi_az = True doesn't actually set an instance to MultiAZ\nMaking a call to a non-multiaz instance with multi_az=True doesn't actually switch the parameter. I assume this is also true for creating one from scratch, but I haven't tested that yet.\n\n", "before_files": [{"content": "# Copyright (c) 2010 Spotify AB\n#\n# Permission is hereby granted, free of charge, to any person obtaining a\n# copy of this software and associated documentation files (the\n# \"Software\"), to deal in the Software without restriction, including\n# without limitation the rights to use, copy, modify, merge, publish, dis-\n# tribute, sublicense, and/or sell copies of the Software, and to permit\n# persons to whom the Software is furnished to do so, subject to the fol-\n# lowing conditions:\n#\n# The above copyright notice and this permission notice shall be included\n# in all copies or substantial portions of the Software.\n#\n# THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS\n# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-\n# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT\n# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,\n# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS\n# IN THE SOFTWARE.\n\n\"\"\"\nRepresents a connection to the EMR service\n\"\"\"\nimport types\n\nimport boto\nfrom boto.ec2.regioninfo import RegionInfo\nfrom boto.emr.emrobject import JobFlow, RunJobFlowResponse\nfrom boto.emr.step import JarStep\nfrom boto.connection import AWSQueryConnection\nfrom boto.exception import EmrResponseError\n\nclass EmrConnection(AWSQueryConnection):\n\n    APIVersion = boto.config.get('Boto', 'emr_version', '2009-03-31')\n    DefaultRegionName = boto.config.get('Boto', 'emr_region_name', 'us-east-1')\n    DefaultRegionEndpoint = boto.config.get('Boto', 'emr_region_endpoint',\n                                            'elasticmapreduce.amazonaws.com')\n    ResponseError = EmrResponseError\n\n    # Constants for AWS Console debugging\n    DebuggingJar = 's3n://us-east-1.elasticmapreduce/libs/script-runner/script-runner.jar'\n    DebuggingArgs = 's3n://us-east-1.elasticmapreduce/libs/state-pusher/0.1/fetch'\n\n    def __init__(self, aws_access_key_id=None, aws_secret_access_key=None,\n                 is_secure=True, port=None, proxy=None, proxy_port=None,\n                 proxy_user=None, proxy_pass=None, debug=0,\n                 https_connection_factory=None, region=None, path='/'):\n        if not region:\n            region = RegionInfo(self, self.DefaultRegionName, self.DefaultRegionEndpoint)\n        self.region = region\n        AWSQueryConnection.__init__(self, aws_access_key_id,\n                                    aws_secret_access_key,\n                                    is_secure, port, proxy, proxy_port,\n                                    proxy_user, proxy_pass,\n                                    self.region.endpoint, debug,\n                                    https_connection_factory, path)\n\n    def _required_auth_capability(self):\n        return ['emr']\n\n    def describe_jobflow(self, jobflow_id):\n        \"\"\"\n        Describes a single Elastic MapReduce job flow\n\n        :type jobflow_id: str\n        :param jobflow_id: The job flow id of interest\n        \"\"\"\n        jobflows = self.describe_jobflows(jobflow_ids=[jobflow_id])\n        if jobflows:\n            return jobflows[0]\n\n    def describe_jobflows(self, states=None, jobflow_ids=None,\n                           created_after=None, created_before=None):\n        \"\"\"\n        Retrieve all the Elastic MapReduce job flows on your account\n\n        :type states: list\n        :param states: A list of strings with job flow states wanted\n\n        :type jobflow_ids: list\n        :param jobflow_ids: A list of job flow IDs\n        :type created_after: datetime\n        :param created_after: Bound on job flow creation time\n\n        :type created_before: datetime\n        :param created_before: Bound on job flow creation time\n        \"\"\"\n        params = {}\n\n        if states:\n            self.build_list_params(params, states, 'JobFlowStates.member')\n        if jobflow_ids:\n            self.build_list_params(params, jobflow_ids, 'JobFlowIds.member')\n        if created_after:\n            params['CreatedAfter'] = created_after.strftime('%Y-%m-%dT%H:%M:%S')\n        if created_before:\n            params['CreatedBefore'] = created_before.strftime('%Y-%m-%dT%H:%M:%S')\n\n        return self.get_list('DescribeJobFlows', params, [('member', JobFlow)])\n\n    def terminate_jobflow(self, jobflow_id):\n        \"\"\"\n        Terminate an Elastic MapReduce job flow\n\n        :type jobflow_id: str\n        :param jobflow_id: A jobflow id \n        \"\"\"\n        self.terminate_jobflows([jobflow_id]) \n\n    def terminate_jobflows(self, jobflow_ids):\n        \"\"\"\n        Terminate an Elastic MapReduce job flow\n\n        :type jobflow_ids: list\n        :param jobflow_ids: A list of job flow IDs\n        \"\"\"\n        params = {}\n        self.build_list_params(params, jobflow_ids, 'JobFlowIds.member')\n        return self.get_status('TerminateJobFlows', params)\n\n    def add_jobflow_steps(self, jobflow_id, steps):\n        \"\"\"\n        Adds steps to a jobflow\n\n        :type jobflow_id: str\n        :param jobflow_id: The job flow id\n        :type steps: list(boto.emr.Step)\n        :param steps: A list of steps to add to the job\n        \"\"\"\n        if type(steps) != types.ListType:\n            steps = [steps]\n        params = {}\n        params['JobFlowId'] = jobflow_id\n\n        # Step args\n        step_args = [self._build_step_args(step) for step in steps]\n        params.update(self._build_step_list(step_args))\n\n        return self.get_object('AddJobFlowSteps', params, RunJobFlowResponse)\n\n    def run_jobflow(self, name, log_uri, ec2_keyname=None, availability_zone=None,\n                    master_instance_type='m1.small',\n                    slave_instance_type='m1.small', num_instances=1,\n                    action_on_failure='TERMINATE_JOB_FLOW', keep_alive=False,\n                    enable_debugging=False,\n                    hadoop_version='0.18',\n                    steps=[],\n                    bootstrap_actions=[]):\n        \"\"\"\n        Runs a job flow\n\n        :type name: str\n        :param name: Name of the job flow\n        :type log_uri: str\n        :param log_uri: URI of the S3 bucket to place logs\n        :type ec2_keyname: str\n        :param ec2_keyname: EC2 key used for the instances\n        :type availability_zone: str\n        :param availability_zone: EC2 availability zone of the cluster\n        :type master_instance_type: str\n        :param master_instance_type: EC2 instance type of the master\n        :type slave_instance_type: str\n        :param slave_instance_type: EC2 instance type of the slave nodes\n        :type num_instances: int\n        :param num_instances: Number of instances in the Hadoop cluster\n        :type action_on_failure: str\n        :param action_on_failure: Action to take if a step terminates\n        :type keep_alive: bool\n        :param keep_alive: Denotes whether the cluster should stay alive upon completion\n        :type enable_debugging: bool\n        :param enable_debugging: Denotes whether AWS console debugging should be enabled.\n        :type steps: list(boto.emr.Step)\n        :param steps: List of steps to add with the job\n\n        :rtype: str\n        :return: The jobflow id\n        \"\"\"\n        params = {}\n        if action_on_failure:\n            params['ActionOnFailure'] = action_on_failure\n        params['Name'] = name\n        params['LogUri'] = log_uri\n\n        # Instance args\n        instance_params = self._build_instance_args(ec2_keyname, availability_zone,\n                                                    master_instance_type, slave_instance_type,\n                                                    num_instances, keep_alive, hadoop_version)\n        params.update(instance_params)\n\n        # Debugging step from EMR API docs\n        if enable_debugging:\n            debugging_step = JarStep(name='Setup Hadoop Debugging',\n                                     action_on_failure='TERMINATE_JOB_FLOW',\n                                     main_class=None,\n                                     jar=self.DebuggingJar,\n                                     step_args=self.DebuggingArgs)\n            steps.insert(0, debugging_step)\n\n        # Step args\n        if steps:\n            step_args = [self._build_step_args(step) for step in steps]\n            params.update(self._build_step_list(step_args))\n\n        if bootstrap_actions:\n            bootstrap_action_args = [self._build_bootstrap_action_args(bootstrap_action) for bootstrap_action in bootstrap_actions]\n            params.update(self._build_bootstrap_action_list(bootstrap_action_args))\n\n        response = self.get_object('RunJobFlow', params, RunJobFlowResponse)\n        return response.jobflowid\n\n    def _build_bootstrap_action_args(self, bootstrap_action):\n        bootstrap_action_params = {}\n        bootstrap_action_params['ScriptBootstrapAction.Path'] = bootstrap_action.path\n\n        try:\n            bootstrap_action_params['Name'] = bootstrap_action.name\n        except AttributeError:\n            pass\n\n        args = bootstrap_action.args()\n        if args:\n            self.build_list_params(bootstrap_action_params, args, 'ScriptBootstrapAction.Args.member')\n\n        return bootstrap_action_params\n\n    def _build_step_args(self, step):\n        step_params = {}\n        step_params['ActionOnFailure'] = step.action_on_failure\n        step_params['HadoopJarStep.Jar'] = step.jar()\n\n        main_class = step.main_class()\n        if main_class:\n            step_params['HadoopJarStep.MainClass'] = main_class\n\n        args = step.args()\n        if args:\n            self.build_list_params(step_params, args, 'HadoopJarStep.Args.member')\n\n        step_params['Name'] = step.name\n        return step_params\n\n    def _build_bootstrap_action_list(self, bootstrap_actions):\n        if type(bootstrap_actions) != types.ListType:\n            bootstrap_actions = [bootstrap_actions]\n\n        params = {}\n        for i, bootstrap_action in enumerate(bootstrap_actions):\n            for key, value in bootstrap_action.iteritems():\n                params['BootstrapActions.member.%s.%s' % (i + 1, key)] = value\n        return params\n\n    def _build_step_list(self, steps):\n        if type(steps) != types.ListType:\n            steps = [steps]\n\n        params = {}\n        for i, step in enumerate(steps):\n            for key, value in step.iteritems():\n                params['Steps.member.%s.%s' % (i+1, key)] = value\n        return params\n\n    def _build_instance_args(self, ec2_keyname, availability_zone, master_instance_type,\n                             slave_instance_type, num_instances, keep_alive, hadoop_version):\n        params = {\n            'Instances.MasterInstanceType' : master_instance_type,\n            'Instances.SlaveInstanceType' : slave_instance_type,\n            'Instances.InstanceCount' : num_instances,\n            'Instances.KeepJobFlowAliveWhenNoSteps' : str(keep_alive).lower(),\n            'Instances.HadoopVersion' : hadoop_version\n        }\n\n        if ec2_keyname:\n            params['Instances.Ec2KeyName'] = ec2_keyname\n        if availability_zone:\n            params['Placement'] = availability_zone\n\n        return params\n\n", "path": "boto/emr/connection.py"}], "after_files": [{"content": "# Copyright (c) 2010 Spotify AB\n#\n# Permission is hereby granted, free of charge, to any person obtaining a\n# copy of this software and associated documentation files (the\n# \"Software\"), to deal in the Software without restriction, including\n# without limitation the rights to use, copy, modify, merge, publish, dis-\n# tribute, sublicense, and/or sell copies of the Software, and to permit\n# persons to whom the Software is furnished to do so, subject to the fol-\n# lowing conditions:\n#\n# The above copyright notice and this permission notice shall be included\n# in all copies or substantial portions of the Software.\n#\n# THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS\n# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-\n# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT\n# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,\n# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS\n# IN THE SOFTWARE.\n\n\"\"\"\nRepresents a connection to the EMR service\n\"\"\"\nimport types\n\nimport boto\nfrom boto.ec2.regioninfo import RegionInfo\nfrom boto.emr.emrobject import JobFlow, RunJobFlowResponse\nfrom boto.emr.step import JarStep\nfrom boto.connection import AWSQueryConnection\nfrom boto.exception import EmrResponseError\n\nclass EmrConnection(AWSQueryConnection):\n\n    APIVersion = boto.config.get('Boto', 'emr_version', '2009-03-31')\n    DefaultRegionName = boto.config.get('Boto', 'emr_region_name', 'us-east-1')\n    DefaultRegionEndpoint = boto.config.get('Boto', 'emr_region_endpoint',\n                                            'elasticmapreduce.amazonaws.com')\n    ResponseError = EmrResponseError\n\n    # Constants for AWS Console debugging\n    DebuggingJar = 's3n://us-east-1.elasticmapreduce/libs/script-runner/script-runner.jar'\n    DebuggingArgs = 's3n://us-east-1.elasticmapreduce/libs/state-pusher/0.1/fetch'\n\n    def __init__(self, aws_access_key_id=None, aws_secret_access_key=None,\n                 is_secure=True, port=None, proxy=None, proxy_port=None,\n                 proxy_user=None, proxy_pass=None, debug=0,\n                 https_connection_factory=None, region=None, path='/'):\n        if not region:\n            region = RegionInfo(self, self.DefaultRegionName, self.DefaultRegionEndpoint)\n        self.region = region\n        AWSQueryConnection.__init__(self, aws_access_key_id,\n                                    aws_secret_access_key,\n                                    is_secure, port, proxy, proxy_port,\n                                    proxy_user, proxy_pass,\n                                    self.region.endpoint, debug,\n                                    https_connection_factory, path)\n\n    def _required_auth_capability(self):\n        return ['emr']\n\n    def describe_jobflow(self, jobflow_id):\n        \"\"\"\n        Describes a single Elastic MapReduce job flow\n\n        :type jobflow_id: str\n        :param jobflow_id: The job flow id of interest\n        \"\"\"\n        jobflows = self.describe_jobflows(jobflow_ids=[jobflow_id])\n        if jobflows:\n            return jobflows[0]\n\n    def describe_jobflows(self, states=None, jobflow_ids=None,\n                           created_after=None, created_before=None):\n        \"\"\"\n        Retrieve all the Elastic MapReduce job flows on your account\n\n        :type states: list\n        :param states: A list of strings with job flow states wanted\n\n        :type jobflow_ids: list\n        :param jobflow_ids: A list of job flow IDs\n        :type created_after: datetime\n        :param created_after: Bound on job flow creation time\n\n        :type created_before: datetime\n        :param created_before: Bound on job flow creation time\n        \"\"\"\n        params = {}\n\n        if states:\n            self.build_list_params(params, states, 'JobFlowStates.member')\n        if jobflow_ids:\n            self.build_list_params(params, jobflow_ids, 'JobFlowIds.member')\n        if created_after:\n            params['CreatedAfter'] = created_after.strftime('%Y-%m-%dT%H:%M:%S')\n        if created_before:\n            params['CreatedBefore'] = created_before.strftime('%Y-%m-%dT%H:%M:%S')\n\n        return self.get_list('DescribeJobFlows', params, [('member', JobFlow)])\n\n    def terminate_jobflow(self, jobflow_id):\n        \"\"\"\n        Terminate an Elastic MapReduce job flow\n\n        :type jobflow_id: str\n        :param jobflow_id: A jobflow id \n        \"\"\"\n        self.terminate_jobflows([jobflow_id]) \n\n    def terminate_jobflows(self, jobflow_ids):\n        \"\"\"\n        Terminate an Elastic MapReduce job flow\n\n        :type jobflow_ids: list\n        :param jobflow_ids: A list of job flow IDs\n        \"\"\"\n        params = {}\n        self.build_list_params(params, jobflow_ids, 'JobFlowIds.member')\n        return self.get_status('TerminateJobFlows', params)\n\n    def add_jobflow_steps(self, jobflow_id, steps):\n        \"\"\"\n        Adds steps to a jobflow\n\n        :type jobflow_id: str\n        :param jobflow_id: The job flow id\n        :type steps: list(boto.emr.Step)\n        :param steps: A list of steps to add to the job\n        \"\"\"\n        if type(steps) != types.ListType:\n            steps = [steps]\n        params = {}\n        params['JobFlowId'] = jobflow_id\n\n        # Step args\n        step_args = [self._build_step_args(step) for step in steps]\n        params.update(self._build_step_list(step_args))\n\n        return self.get_object('AddJobFlowSteps', params, RunJobFlowResponse)\n\n    def run_jobflow(self, name, log_uri, ec2_keyname=None, availability_zone=None,\n                    master_instance_type='m1.small',\n                    slave_instance_type='m1.small', num_instances=1,\n                    action_on_failure='TERMINATE_JOB_FLOW', keep_alive=False,\n                    enable_debugging=False,\n                    hadoop_version='0.18',\n                    steps=[],\n                    bootstrap_actions=[]):\n        \"\"\"\n        Runs a job flow\n\n        :type name: str\n        :param name: Name of the job flow\n        :type log_uri: str\n        :param log_uri: URI of the S3 bucket to place logs\n        :type ec2_keyname: str\n        :param ec2_keyname: EC2 key used for the instances\n        :type availability_zone: str\n        :param availability_zone: EC2 availability zone of the cluster\n        :type master_instance_type: str\n        :param master_instance_type: EC2 instance type of the master\n        :type slave_instance_type: str\n        :param slave_instance_type: EC2 instance type of the slave nodes\n        :type num_instances: int\n        :param num_instances: Number of instances in the Hadoop cluster\n        :type action_on_failure: str\n        :param action_on_failure: Action to take if a step terminates\n        :type keep_alive: bool\n        :param keep_alive: Denotes whether the cluster should stay alive upon completion\n        :type enable_debugging: bool\n        :param enable_debugging: Denotes whether AWS console debugging should be enabled.\n        :type steps: list(boto.emr.Step)\n        :param steps: List of steps to add with the job\n\n        :rtype: str\n        :return: The jobflow id\n        \"\"\"\n        params = {}\n        if action_on_failure:\n            params['ActionOnFailure'] = action_on_failure\n        params['Name'] = name\n        params['LogUri'] = log_uri\n\n        # Instance args\n        instance_params = self._build_instance_args(ec2_keyname, availability_zone,\n                                                    master_instance_type, slave_instance_type,\n                                                    num_instances, keep_alive, hadoop_version)\n        params.update(instance_params)\n\n        # Debugging step from EMR API docs\n        if enable_debugging:\n            debugging_step = JarStep(name='Setup Hadoop Debugging',\n                                     action_on_failure='TERMINATE_JOB_FLOW',\n                                     main_class=None,\n                                     jar=self.DebuggingJar,\n                                     step_args=self.DebuggingArgs)\n            steps.insert(0, debugging_step)\n\n        # Step args\n        if steps:\n            step_args = [self._build_step_args(step) for step in steps]\n            params.update(self._build_step_list(step_args))\n\n        if bootstrap_actions:\n            bootstrap_action_args = [self._build_bootstrap_action_args(bootstrap_action) for bootstrap_action in bootstrap_actions]\n            params.update(self._build_bootstrap_action_list(bootstrap_action_args))\n\n        response = self.get_object('RunJobFlow', params, RunJobFlowResponse)\n        return response.jobflowid\n\n    def _build_bootstrap_action_args(self, bootstrap_action):\n        bootstrap_action_params = {}\n        bootstrap_action_params['ScriptBootstrapAction.Path'] = bootstrap_action.path\n\n        try:\n            bootstrap_action_params['Name'] = bootstrap_action.name\n        except AttributeError:\n            pass\n\n        args = bootstrap_action.args()\n        if args:\n            self.build_list_params(bootstrap_action_params, args, 'ScriptBootstrapAction.Args.member')\n\n        return bootstrap_action_params\n\n    def _build_step_args(self, step):\n        step_params = {}\n        step_params['ActionOnFailure'] = step.action_on_failure\n        step_params['HadoopJarStep.Jar'] = step.jar()\n\n        main_class = step.main_class()\n        if main_class:\n            step_params['HadoopJarStep.MainClass'] = main_class\n\n        args = step.args()\n        if args:\n            self.build_list_params(step_params, args, 'HadoopJarStep.Args.member')\n\n        step_params['Name'] = step.name\n        return step_params\n\n    def _build_bootstrap_action_list(self, bootstrap_actions):\n        if type(bootstrap_actions) != types.ListType:\n            bootstrap_actions = [bootstrap_actions]\n\n        params = {}\n        for i, bootstrap_action in enumerate(bootstrap_actions):\n            for key, value in bootstrap_action.iteritems():\n                params['BootstrapActions.member.%s.%s' % (i + 1, key)] = value\n        return params\n\n    def _build_step_list(self, steps):\n        if type(steps) != types.ListType:\n            steps = [steps]\n\n        params = {}\n        for i, step in enumerate(steps):\n            for key, value in step.iteritems():\n                params['Steps.member.%s.%s' % (i+1, key)] = value\n        return params\n\n    def _build_instance_args(self, ec2_keyname, availability_zone, master_instance_type,\n                             slave_instance_type, num_instances, keep_alive, hadoop_version):\n        params = {\n            'Instances.MasterInstanceType' : master_instance_type,\n            'Instances.SlaveInstanceType' : slave_instance_type,\n            'Instances.InstanceCount' : num_instances,\n            'Instances.KeepJobFlowAliveWhenNoSteps' : str(keep_alive).lower(),\n            'Instances.HadoopVersion' : hadoop_version\n        }\n\n        if ec2_keyname:\n            params['Instances.Ec2KeyName'] = ec2_keyname\n        if availability_zone:\n            params['Placement.AvailabilityZone'] = availability_zone\n\n        return params\n\n", "path": "boto/emr/connection.py"}]}