Optionally address streams by ARN instead of name (#39)

Author: amotl

README.md

@@ -17,7 +17,7 @@ pip install async-kinesis
   - deadlock + reallocation of shards if checkpoint fails to heartbeat within "session_timeout"
 - processors (aggregator + serializer)
     - json line delimited, msgpack
-
+- Address Kinesis streams by name or [ARN]
 
 See [docs/design](./docs/DESIGN.md) for more details.
 See [docs/yetanother](docs/YETANOTHER.md) as to why reinvent the wheel.
@@ -31,6 +31,11 @@ AWS_ACCESS_KEY_ID
 AWS_SECRET_ACCESS_KEY
 ```
 
+## Stream addressing
+
+The `stream_name` argument to consumer and producer accepts either the [StreamName]
+like `test`, or a full [StreamARN] like `arn:aws:kinesis:eu-central-1:842404475894:stream/test`.
+
 ## Producer
 
     from kinesis import Producer
@@ -206,3 +211,6 @@ TESTING_USE_AWS_KINESIS=1
 Note you can ignore these tests if submitting PR unless core batching/processing behaviour is being changed.
 
 
+[ARN]: https://docs.aws.amazon.com/IAM/latest/UserGuide/reference-arns.html
+[StreamARN]: https://docs.aws.amazon.com/kinesis/latest/APIReference/API_StreamDescription.html#Streams-Type-StreamDescription-StreamARN
+[StreamName]: https://docs.aws.amazon.com/kinesis/latest/APIReference/API_StreamDescription.html#Streams-Type-StreamDescription-StreamName

kinesis/base.py

@@ -82,6 +82,19 @@ async def __aexit__(self, exc_type, exc, tb):
         await self.close()
         await self.client.__aexit__(exc_type, exc, tb)
 
+    @property
+    def address(self):
+        """
+        Return address of stream, either as StreamName or StreamARN, when applicable.
+
+        https://docs.aws.amazon.com/kinesis/latest/APIReference/API_StreamDescription.html#Streams-Type-StreamDescription-StreamName
+        https://docs.aws.amazon.com/kinesis/latest/APIReference/API_StreamDescription.html#Streams-Type-StreamDescription-StreamARN
+        """
+        if self.stream_name.startswith("arn:"):
+            return {"StreamARN": self.stream_name}
+        else:
+            return {"StreamName": self.stream_name}
+
     async def get_client(self):
 
         # Note: max_attempts = 0
@@ -101,7 +114,7 @@ async def get_client(self):
     async def get_stream_description(self):
 
         try:
-            return (await self.client.describe_stream(StreamName=self.stream_name))[
+            return (await self.client.describe_stream(**self.address))[
                 "StreamDescription"
             ]
         except ClientError as err:
@@ -240,7 +253,7 @@ async def _create_stream(self, ignore_exists=True):
 
         try:
             await self.client.create_stream(
-                StreamName=self.stream_name, ShardCount=self.create_stream_shards
+                **self.address, ShardCount=self.create_stream_shards
             )
         except ClientError as err:
             code = err.response["Error"]["Code"]

kinesis/consumer.py

@@ -353,12 +353,12 @@ async def get_shard_iterator(self, shard_id, last_sequence_number=None):
         )
 
         params = {
-            "StreamName": self.stream_name,
             "ShardId": shard_id,
             "ShardIteratorType": "AFTER_SEQUENCE_NUMBER"
             if last_sequence_number
             else self.iterator_type,
         }
+        params.update(self.address)
 
         if last_sequence_number:
             params["StartingSequenceNumber"] = last_sequence_number

kinesis/producer.py

@@ -279,7 +279,7 @@ async def _push_kinesis(self, items):
                         }
                         for item in items
                     ],
-                    StreamName=self.stream_name,
+                    **self.address,
                 )
 
                 log.info(