Repository: aws/chalice

Issue:
Using on_sqs_message with a CDK created queue
I'm playing with the "[CDK] Rest API with a DynamoDB table" demo project and am trying to add an SQS queue+handler to it.

I have added the queue like so:
   ```
 def _create_queue(self) -> sqs.Queue:
        the_queue = sqs.Queue(
            self, 'the_queue',
            delivery_delay=cdk.Duration.seconds(15),
        )
        cdk.CfnOutput(self, 'QueueName', value=the_queue.queue_name)
        return the_queue
```

And added the handler like so:
```
@app.on_sqs_message(queue=os.environ.get('THE_QUEUE', ''))
def the_queue_handler(event):
    app.log.info("Event: %s", event.to_dict())
    for record in event:
        app.log.info("Message body: %s", record.body)
```

The queue appears to be created correctly, but registering the EventSourceMapping fails:
```
12:19:23 pm | CREATE_FAILED        | AWS::Lambda::EventSourceMapping | HandlerHandlerSqsEventSource
Invalid request provided: Invalid event source arn: The Relative Id must be non-empty: arn:aws:sqs:ap-southeast-2:341354347128: (Service: Lambda, Status Co
de: 400, Request ID: aca69971-9c5f-4529-8182-738f1e4d044a, Extended Request ID: null)
```

Have I done something wrong?



Discussion:

Patch:
diff --git a/.changes/next-release/41104421952-feature-SQS-66900.json b/.changes/next-release/41104421952-feature-SQS-66900.json
new file mode 100644
index 0000000..0d2c63d
--- /dev/null
+++ b/.changes/next-release/41104421952-feature-SQS-66900.json
@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "SQS",
+  "description": "Add queue_arn parameter to enable CDK integration with SQS event handler (#1681)"
+}
diff --git a/chalice/app.py b/chalice/app.py
index 81e9226..8f84502 100644
--- a/chalice/app.py
+++ b/chalice/app.py
@@ -720,11 +720,16 @@ class DecoratorAPI(object):
             registration_kwargs={'topic': topic}
         )
 
-    def on_sqs_message(self, queue, batch_size=1, name=None):
+    def on_sqs_message(self, queue=None, batch_size=1,
+                       name=None, queue_arn=None):
         return self._create_registration_function(
             handler_type='on_sqs_message',
             name=name,
-            registration_kwargs={'queue': queue, 'batch_size': batch_size}
+            registration_kwargs={
+                'queue': queue,
+                'queue_arn': queue_arn,
+                'batch_size': batch_size,
+            }
         )
 
     def on_cw_event(self, event_pattern, name=None):
@@ -976,10 +981,18 @@ class _HandlerRegistration(object):
         self.event_sources.append(sns_config)
 
     def _register_on_sqs_message(self, name, handler_string, kwargs, **unused):
+        queue = kwargs.get('queue')
+        queue_arn = kwargs.get('queue_arn')
+        if not queue and not queue_arn:
+            raise ValueError(
+                "Must provide either `queue` or `queue_arn` to the "
+                "`on_sqs_message` decorator."
+            )
         sqs_config = SQSEventConfig(
             name=name,
             handler_string=handler_string,
-            queue=kwargs['queue'],
+            queue=queue,
+            queue_arn=queue_arn,
             batch_size=kwargs['batch_size'],
         )
         self.event_sources.append(sqs_config)
@@ -1442,9 +1455,10 @@ class SNSEventConfig(BaseEventSourceConfig):
 
 
 class SQSEventConfig(BaseEventSourceConfig):
-    def __init__(self, name, handler_string, queue, batch_size):
+    def __init__(self, name, handler_string, queue, queue_arn, batch_size):
         super(SQSEventConfig, self).__init__(name, handler_string)
         self.queue = queue
+        self.queue_arn = queue_arn
         self.batch_size = batch_size
 
 
diff --git a/chalice/app.pyi b/chalice/app.pyi
index 17daf52..b922533 100644
--- a/chalice/app.pyi
+++ b/chalice/app.pyi
@@ -354,11 +354,13 @@ class SNSEventConfig(BaseEventSourceConfig):
 
 
 class SQSEventConfig(BaseEventSourceConfig):
-    queue = ... # type: str
+    queue = ... # type: Optional[str]
+    queue_arn = ... # type: Optional[str]
     batch_size = ... # type: int
 
     def __init__(
-        self, name: str, handler_string: str, queue: str, batch_size: int
+        self, name: str, handler_string: str, queue: Optional[str],
+        queue_arn: Optional[str], batch_size: int
     ) -> None: ...
 
 
diff --git a/chalice/deploy/appgraph.py b/chalice/deploy/appgraph.py
index 150a293..7606631 100644
--- a/chalice/deploy/appgraph.py
+++ b/chalice/deploy/appgraph.py
@@ -560,9 +560,14 @@ class ApplicationGraphBuilder(object):
             handler_name=sqs_config.handler_string, stage_name=stage_name
         )
         resource_name = sqs_config.name + '-sqs-event-source'
+        queue = ''  # type: Union[str, models.QueueARN]
+        if sqs_config.queue_arn is not None:
+            queue = models.QueueARN(arn=sqs_config.queue_arn)
+        elif sqs_config.queue is not None:
+            queue = sqs_config.queue
         sqs_event_source = models.SQSEventSource(
             resource_name=resource_name,
-            queue=sqs_config.queue,
+            queue=queue,
             batch_size=sqs_config.batch_size,
             lambda_function=lambda_function,
         )
diff --git a/chalice/deploy/models.py b/chalice/deploy/models.py
index 8f52be8..0c4dfba 100644
--- a/chalice/deploy/models.py
+++ b/chalice/deploy/models.py
@@ -329,10 +329,22 @@ class SNSLambdaSubscription(FunctionEventSubscriber):
     topic = attrib()            # type: str
 
 
+@attrs
+class QueueARN(object):
+    arn = attrib()  # type: str
+
+    @property
+    def queue_name(self):
+        # type: () -> str
+        # Pylint 2.x validates this correctly, but for py27, we have to
+        # use Pylint 1.x which doesn't support attrs.
+        return self.arn.rpartition(':')[2]  # pylint: disable=no-member
+
+
 @attrs
 class SQSEventSource(FunctionEventSubscriber):
     resource_type = 'sqs_event'
-    queue = attrib()            # type: str
+    queue = attrib()            # type: Union[str, QueueARN]
     batch_size = attrib()       # type: int
 
 
diff --git a/chalice/deploy/planner.py b/chalice/deploy/planner.py
index 63c6597..d9a9de9 100644
--- a/chalice/deploy/planner.py
+++ b/chalice/deploy/planner.py
@@ -91,9 +91,13 @@ class RemoteState(object):
                 resource.resource_name)
         except ValueError:
             return False
+        if isinstance(resource.queue, models.QueueARN):
+            resource_name = resource.queue.queue_name
+        else:
+            resource_name = resource.queue
         return self._client.verify_event_source_current(
             event_uuid=deployed_values['event_uuid'],
-            resource_name=resource.queue,
+            resource_name=resource_name,
             service_name='sqs',
             function_arn=deployed_values['lambda_arn'],
         )
@@ -694,18 +698,29 @@ class PlanStage(object):
         function_arn = Variable(
             '%s_lambda_arn' % resource.lambda_function.resource_name
         )
-        instruction_for_queue_arn = self._arn_parse_instructions(function_arn)
-        instruction_for_queue_arn.append(
-            models.StoreValue(
-                name=queue_arn_varname,
-                value=StringFormat(
-                    'arn:{partition}:sqs:{region_name}:{account_id}:%s' % (
-                        resource.queue
+        if not isinstance(resource.queue, models.QueueARN):
+            instruction_for_queue_arn = self._arn_parse_instructions(
+                function_arn)
+            instruction_for_queue_arn.append(
+                models.StoreValue(
+                    name=queue_arn_varname,
+                    value=StringFormat(
+                        'arn:{partition}:sqs:{region_name}:{account_id}:%s' % (
+                            resource.queue
+                        ),
+                        ['partition', 'region_name', 'account_id'],
                     ),
-                    ['partition', 'region_name', 'account_id'],
-                ),
+                )
             )
-        )
+            queue_name = resource.queue
+        else:
+            instruction_for_queue_arn = [
+                models.StoreValue(
+                    name=queue_arn_varname,
+                    value=resource.queue.arn,
+                )
+            ]
+            queue_name = resource.queue.queue_name
         if self._remote_state.resource_exists(resource):
             deployed = self._remote_state.resource_deployed_values(resource)
             uuid = deployed['event_uuid']
@@ -719,7 +734,7 @@ class PlanStage(object):
                 'sqs_event', resource.resource_name, {
                     'queue_arn': deployed['queue_arn'],
                     'event_uuid': uuid,
-                    'queue': resource.queue,
+                    'queue': queue_name,
                     'lambda_arn': deployed['lambda_arn'],
                 }
             )
@@ -737,7 +752,7 @@ class PlanStage(object):
             'sqs_event', resource.resource_name, {
                 'queue_arn': Variable(queue_arn_varname),
                 'event_uuid': Variable(uuid_varname),
-                'queue': resource.queue,
+                'queue': queue_name,
                 'lambda_arn': Variable(function_arn.name)
             }
         )
diff --git a/chalice/deploy/validate.py b/chalice/deploy/validate.py
index 0a53c95..003e0f2 100644
--- a/chalice/deploy/validate.py
+++ b/chalice/deploy/validate.py
@@ -241,21 +241,24 @@ def validate_sqs_configuration(chalice_app):
     for event in chalice_app.event_sources:
         if not isinstance(event, app.SQSEventConfig):
             continue
-        if not _is_valid_queue_name(event.queue):
+        if not _is_valid_queue_name(event.queue, event.queue_arn):
             raise ValueError("The 'queue' parameter for the "
                              "'@app.on_sqs_message()' handler must be the "
                              "name of the queue, not the queue URL or the "
                              "queue ARN.  Invalid value: %s" % event.queue)
 
 
-def _is_valid_queue_name(queue_name):
-    # type: (str) -> bool
-    if queue_name.startswith(('https:', 'arn:')):
+def _is_valid_queue_name(queue_name, queue_arn):
+    # type: (Optional[str], Optional[str]) -> bool
+    # The mutually exclusiveness is verified in the on_sqs_message decorator.
+    if queue_name is not None and queue_name.startswith(('https:', 'arn:')):
+        return False
+    if queue_arn is not None and not queue_arn.startswith('arn:'):
         return False
     # We're not validating that the queue has only valid chars because SQS
     # won't let you create a queue with that name in the first place.  We just
     # want to detect the case where a user puts the queue URL/ARN instead of
-    # the name.
+    # the name for the queue_name.
     return True
 
 
diff --git a/chalice/package.py b/chalice/package.py
index cd372ca..6d3e12a 100644
--- a/chalice/package.py
+++ b/chalice/package.py
@@ -629,19 +629,21 @@ class SAMTemplateGenerator(TemplateGenerator):
         function_cfn_name = to_cfn_resource_name(
             resource.lambda_function.resource_name)
         function_cfn = template['Resources'][function_cfn_name]
-        sns_cfn_name = self._register_cfn_resource_name(
+        sqs_cfn_name = self._register_cfn_resource_name(
             resource.resource_name)
+        queue = ''  # type: Union[str, Dict[str, Any]]
+        if isinstance(resource.queue, models.QueueARN):
+            queue = resource.queue.arn
+        else:
+            queue = {
+                'Fn::Sub': ('arn:${AWS::Partition}:sqs:${AWS::Region}'
+                            ':${AWS::AccountId}:%s' % resource.queue)
+            }
         function_cfn['Properties']['Events'] = {
-            sns_cfn_name: {
+            sqs_cfn_name: {
                 'Type': 'SQS',
                 'Properties': {
-                    'Queue': {
-                        'Fn::Sub': (
-                            'arn:${AWS::Partition}:sqs:${AWS::Region}'
-                            ':${AWS::AccountId}:%s' %
-                            resource.queue
-                        )
-                    },
+                    'Queue': queue,
                     'BatchSize': resource.batch_size,
                 }
             }
@@ -901,12 +903,17 @@ class TerraformGenerator(TemplateGenerator):
 
     def _generate_sqseventsource(self, resource, template):
         # type: (models.SQSEventSource, Dict[str, Any]) -> None
-        template['resource'].setdefault('aws_lambda_event_source_mapping', {})[
-            resource.resource_name] = {
-            'event_source_arn': self._arnref(
+        if isinstance(resource.queue, models.QueueARN):
+            event_source_arn = resource.queue.arn
+        else:
+            event_source_arn = self._arnref(
                 "arn:%(partition)s:sqs:%(region)s"
                 ":%(account_id)s:%(queue)s",
-                queue=resource.queue),
+                queue=resource.queue
+            )
+        template['resource'].setdefault('aws_lambda_event_source_mapping', {})[
+            resource.resource_name] = {
+            'event_source_arn': event_source_arn,
             'batch_size': resource.batch_size,
             'function_name': self._fref(resource.lambda_function)
         }
diff --git a/docs/source/api.rst b/docs/source/api.rst
index b5ea118..7fc6f6a 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -282,7 +282,7 @@ Chalice
         entire lambda function name.  This parameter is optional.  If it is
         not provided, the name of the python function will be used.
 
-   .. method:: on_sqs_message(queue, batch_size=1, name=None)
+   .. method:: on_sqs_message(queue, batch_size=1, name=None, queue_arn=None)
 
       Create a lambda function and configure it to be automatically invoked
       whenever a message is published to the specified SQS queue.
@@ -329,6 +329,12 @@ Chalice
         entire lambda function name.  This parameter is optional.  If it is
         not provided, the name of the python function will be used.
 
+      :param queue_arn: The ARN of the SQS queue you want to subscribe to.
+        This argument is mutually exclusive with the ``queue`` parameter.
+        This is useful if you already know the exact ARN or when integrating
+        with the AWS CDK to create your SQS queue.
+
+
    .. method:: on_kinesis_record(stream, batch_size=100, starting_position='LATEST', name=None)
 
       Create a lambda function and configure it to be automatically invoked


Test patch:
diff --git a/tests/unit/deploy/test_appgraph.py b/tests/unit/deploy/test_appgraph.py
index 004db16..f7536b6 100644
--- a/tests/unit/deploy/test_appgraph.py
+++ b/tests/unit/deploy/test_appgraph.py
@@ -526,6 +526,22 @@ class TestApplicationGraphBuilder(object):
         assert lambda_function.resource_name == 'handler'
         assert lambda_function.handler == 'app.handler'
 
+    def test_can_create_sqs_handler_with_queue_arn(self, sample_sqs_event_app):
+        @sample_sqs_event_app.on_sqs_message(queue_arn='arn:my:queue')
+        def new_handler(event):
+            pass
+
+        config = self.create_config(sample_sqs_event_app,
+                                    app_name='sqs-event-app',
+                                    autogen_policy=True)
+        builder = ApplicationGraphBuilder()
+        application = builder.build(config, stage_name='dev')
+        sqs_event = application.resources[1]
+        assert sqs_event.queue == models.QueueARN(arn='arn:my:queue')
+        lambda_function = sqs_event.lambda_function
+        assert lambda_function.resource_name == 'new_handler'
+        assert lambda_function.handler == 'app.new_handler'
+
     def test_can_create_kinesis_event_handler(self, sample_kinesis_event_app):
         config = self.create_config(sample_kinesis_event_app,
                                     app_name='kinesis-event-app',
diff --git a/tests/unit/deploy/test_planner.py b/tests/unit/deploy/test_planner.py
index 5bf8467..54987d7 100644
--- a/tests/unit/deploy/test_planner.py
+++ b/tests/unit/deploy/test_planner.py
@@ -1778,6 +1778,79 @@ class TestPlanSQSSubscription(BasePlannerTests):
             }
         )
 
+    def test_sqs_event_supports_queue_arn(self):
+        function = create_function_resource('function_name')
+        sqs_event_source = models.SQSEventSource(
+            resource_name='function_name-sqs-event-source',
+            queue=models.QueueARN(arn='arn:us-west-2:myqueue'),
+            batch_size=10,
+            lambda_function=function
+        )
+        plan = self.determine_plan(sqs_event_source)
+        assert plan[0] == models.StoreValue(
+            name='function_name-sqs-event-source_queue_arn',
+            value='arn:us-west-2:myqueue',
+        )
+        assert plan[1] == models.APICall(
+            method_name='create_lambda_event_source',
+            params={
+                'event_source_arn': Variable(
+                    "function_name-sqs-event-source_queue_arn"
+                ),
+                'batch_size': 10,
+                'function_name': Variable("function_name_lambda_arn")
+            },
+            output_var='function_name-sqs-event-source_uuid'
+        )
+        self.assert_recorded_values(
+            plan, 'sqs_event', 'function_name-sqs-event-source', {
+                'queue_arn': Variable(
+                    'function_name-sqs-event-source_queue_arn'),
+                'event_uuid': Variable(
+                    'function_name-sqs-event-source_uuid'),
+                'queue': 'myqueue',
+                'lambda_arn': Variable(
+                    'function_name_lambda_arn')
+            }
+        )
+
+    def test_can_update_sqs_event_with_queue_arn(self):
+        function = create_function_resource('function_name')
+        sqs_event_source = models.SQSEventSource(
+            resource_name='function_name-sqs-event-source',
+            queue=models.QueueARN(arn='arn:sqs:myqueue'),
+            batch_size=10,
+            lambda_function=function
+        )
+        self.remote_state.declare_resource_exists(
+            sqs_event_source,
+            queue='myqueue',
+            queue_arn='arn:sqs:myqueue',
+            resource_type='sqs_event',
+            lambda_arn='arn:lambda',
+            event_uuid='my-uuid',
+        )
+        plan = self.determine_plan(sqs_event_source)
+        assert plan[0] == models.StoreValue(
+            name='function_name-sqs-event-source_queue_arn',
+            value='arn:sqs:myqueue',
+        )
+        assert plan[1] == models.APICall(
+            method_name='update_lambda_event_source',
+            params={
+                'event_uuid': 'my-uuid',
+                'batch_size': 10,
+            },
+        )
+        self.assert_recorded_values(
+            plan, 'sqs_event', 'function_name-sqs-event-source', {
+                'queue_arn': 'arn:sqs:myqueue',
+                'event_uuid': 'my-uuid',
+                'queue': 'myqueue',
+                'lambda_arn': 'arn:lambda'
+            }
+        )
+
     def test_sqs_event_source_exists_updates_batch_size(self):
         function = create_function_resource('function_name')
         sqs_event_source = models.SQSEventSource(
diff --git a/tests/unit/deploy/test_validate.py b/tests/unit/deploy/test_validate.py
index d2f73d4..7be913c 100644
--- a/tests/unit/deploy/test_validate.py
+++ b/tests/unit/deploy/test_validate.py
@@ -333,6 +333,27 @@ def test_validate_sqs_queue_name(sample_app):
         validate_configuration(config)
 
 
+def test_can_use_queue_arn(sample_app):
+
+    @sample_app.on_sqs_message(queue_arn='arn:sqs:...:myqueue')
+    def handler(event):
+        pass
+
+    config = Config.create(chalice_app=sample_app)
+    assert validate_configuration(config) is None
+
+
+def test_queue_arn_must_be_arn(sample_app):
+    @sample_app.on_sqs_message(
+        queue_arn='https://sqs.us-west-2.amazonaws.com/12345/myqueue')
+    def handler(event):
+        pass
+
+    config = Config.create(chalice_app=sample_app)
+    with pytest.raises(ValueError):
+        validate_configuration(config)
+
+
 def test_validate_environment_variables_value_type_not_str(sample_app):
     config = Config.create(chalice_app=sample_app,
                            environment_variables={"ENV_KEY": 1})
diff --git a/tests/unit/test_package.py b/tests/unit/test_package.py
index 99e0cdd..fd3c89f 100644
--- a/tests/unit/test_package.py
+++ b/tests/unit/test_package.py
@@ -729,6 +729,25 @@ class TestTerraformTemplate(TemplateTestBase):
                    'batch_size': 5
                }
 
+    def test_sqs_arn_does_not_use_fn_sub(self, sample_app):
+        @sample_app.on_sqs_message(queue_arn='arn:foo:bar', batch_size=5)
+        def handler(event):
+            pass
+
+        config = Config.create(chalice_app=sample_app,
+                               project_dir='.',
+                               app_name='sample_app',
+                               api_gateway_stage='api')
+        template = self.generate_template(config)
+
+        assert template['resource'][
+                   'aws_lambda_event_source_mapping'][
+                   'handler-sqs-event-source'] == {
+                   'event_source_arn': 'arn:foo:bar',
+                   'function_name': '${aws_lambda_function.handler.arn}',
+                   'batch_size': 5
+               }
+
     def test_can_package_kinesis_handler(self, sample_app):
         @sample_app.on_kinesis_record(stream='mystream', batch_size=5,
                                       starting_position='TRIM_HORIZON')
@@ -1521,6 +1540,26 @@ class TestSAMTemplate(TemplateTestBase):
             }
         }
 
+    def test_sqs_arn_does_not_use_fn_sub(self, sample_app):
+        @sample_app.on_sqs_message(queue_arn='arn:foo:bar', batch_size=5)
+        def handler(event):
+            pass
+
+        config = Config.create(chalice_app=sample_app,
+                               project_dir='.',
+                               api_gateway_stage='api')
+        template = self.generate_template(config)
+        sqs_handler = template['Resources']['Handler']
+        assert sqs_handler['Properties']['Events'] == {
+            'HandlerSqsEventSource': {
+                'Type': 'SQS',
+                'Properties': {
+                    'Queue': 'arn:foo:bar',
+                    'BatchSize': 5,
+                },
+            }
+        }
+
     def test_can_package_kinesis_handler(self, sample_app):
         @sample_app.on_kinesis_record(stream='mystream', batch_size=5)
         def handler(event):