Merge pull request #1 from matthewhanson/develop

publish 0.0.3

Author: matthewhanson

CHANGELOG.md

@@ -6,9 +6,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+## [v0.0.3] = 2019-11-22
+
+### Added
+- s3.latest_inventory iterator function added for looping through matching files in an s3 inventory
+- s3.get_presigned_url for generating a presigned URL for s3....does not use boto3
+
 ## [v0.0.2] - 2019-10-08
 
 Initial Release
 
 [Unreleased]: https://github.com/matthewhanson/boto3-utils/compare/master...develop
+[v0.0.3]: https://github.com/matthewhanson/boto3-utils/compare/0.0.2...0.0.3
 [v0.0.2]: https://github.com/matthewhanson/boto3-utils/tree/0.0.2

boto3utils/s3.py

@@ -1,12 +1,14 @@
 import boto3
 import json
+import hashlib
+import hmac
 import logging
+import os
 
 import os.path as op
 
 from botocore.exceptions import ClientError
 from datetime import datetime, timedelta
-from dateutil.parser import parse
 from gzip import GzipFile
 from io import BytesIO
 from os import makedirs, getenv
@@ -85,18 +87,22 @@ def download(uri, path=''):
     return fout
 
 
-def read_json(url):
-    """
-    Download object from S3 as JSON
-    """
+def read(url):
+    """ Read object from s3 """
     parts = urlparse(url)
     response = s3.get_object(Bucket=parts['bucket'], Key=parts['key'])
     body = response['Body'].read()
     if op.splitext(parts['key'])[1] == '.gz':
         body = GzipFile(None, 'rb', fileobj=BytesIO(body)).read()
-    return json.loads(body.decode('utf-8'))
+    return body.decode('utf-8')
 
 
+def read_json(url):
+    """ Download object from S3 as JSON """
+    return json.loads(read(url))
+
+
+# function derived from https://alexwlchan.net/2018/01/listing-s3-keys-redux/
 def find(url, suffix=''):
     """
     Generate objects in an S3 bucket.
@@ -132,3 +138,125 @@ def find(url, suffix=''):
             kwargs['ContinuationToken'] = resp['NextContinuationToken']
         except KeyError:
             break
+
+
+def latest_inventory(url, prefix=None, suffix=None, start_date=None, end_date=None, datetime_key='LastModifiedDate'):
+    """ Return generator function for objects in Bucket with suffix (all files if suffix=None) """
+    parts = urlparse(url)
+    # get latest manifest file
+    today = datetime.now()
+    manifest_key = None
+    for dt in [today, today - timedelta(1)]:
+        _key = op.join(parts['key'], dt.strftime('%Y-%m-%d'))
+        _url = 's3://%s/%s' % (parts['bucket'], _key)
+        keys = [k for k in find(_url, suffix='manifest.json')]
+        if len(keys) == 1:
+            manifest_key = keys[0]
+            break
+    # read through latest manifest looking for matches
+    if manifest_key:
+        _url = 's3://%s/%s' % (parts['bucket'], manifest_key)
+        manifest = read_json(_url)
+        # get file schema
+        keys = [str(key).strip() for key in manifest['fileSchema'].split(',')]
+
+        logger.info('Getting latest inventory from %s' % url)
+        counter = 0
+        for f in manifest.get('files', []):
+            _url = 's3://%s/%s' % (parts['bucket'], f['key'])
+            inv = read(_url).split('\n')
+            for line in inv:
+                counter += 1
+                if counter % 100000 == 0:
+                    logger.debug('%s: Scanned %s records' % (datetime.now(), str(counter)))                
+                info = {keys[i]: v for i, v in  enumerate(line.replace('"', '').split(','))}
+                if 'Key' not in info:
+                    continue
+                # skip to next if last modified date not between start_date and end_date
+                dt = datetime.strptime(info[datetime_key], "%Y-%m-%dT%H:%M:%S.%fZ").date()
+                if (start_date is not None and dt < start_date) or (end_date is not None and dt > end_date):
+                    continue
+                if prefix is not None:
+                    # if path doesn't match provided prefix skip to next record
+                    if info['Key'][:len(prefix)] != prefix:
+                        continue
+                if suffix is None or info['Key'].endswith(suffix):
+                    if 'Bucket' in keys and 'Key' in keys:
+                        info['url'] = 's3://%s/%s' % (info['Bucket'], info['Key']) 
+                    yield info
+
+
+def get_presigned_url(url, aws_region=None,
+                      rtype='GET', public=False, requester_pays=False, content_type=None):
+    """ Get presigned URL """
+    access_key = os.environ.get('AWS_BUCKET_ACCESS_KEY_ID', os.environ.get('AWS_ACCESS_KEY_ID'))
+    secret_key = os.environ.get('AWS_BUCKET_SECRET_ACCESS_KEY', os.environ.get('AWS_SECRET_ACCESS_KEY'))
+    region = os.environ.get('AWS_BUCKET_REGION', os.environ.get('AWS_REGION', 'eu-central-1'))
+    if aws_region is not None:
+        region = aws_region
+    if access_key is None or secret_key is None:
+        # if credentials not provided, just try to download without signed URL
+        logger.debug('Not using signed URL for %s' % url)
+        return url, None
+
+    parts = urlparse(url)
+    bucket = parts['bucket']
+    key = parts['key']
+
+    service = 's3'
+    host = '%s.%s.amazonaws.com' % (bucket, service)
+    request_parameters = ''
+
+    # Key derivation functions. See:
+    # http://docs.aws.amazon.com/general/latest/gr/signature-v4-examples.html#signature-v4-examples-python
+    def sign(key, msg):
+        return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
+
+    def getSignatureKey(key, dateStamp, regionName, serviceName):
+        kDate = sign(('AWS4' + key).encode('utf-8'), dateStamp)
+        kRegion = sign(kDate, regionName)
+        kService = sign(kRegion, serviceName)
+        kSigning = sign(kService, 'aws4_request')
+        return kSigning
+ 
+    # Create a date for headers and the credential string
+    t = datetime.utcnow()
+    amzdate = t.strftime('%Y%m%dT%H%M%SZ')
+    datestamp = t.strftime('%Y%m%d') # Date w/o time, used in credential scope
+
+    # create signed request and headers
+    canonical_uri = '/' + key
+    canonical_querystring = request_parameters
+
+    payload_hash = 'UNSIGNED-PAYLOAD'
+    headers = {
+        'host': host,
+        'x-amz-content-sha256': payload_hash,
+        'x-amz-date': amzdate
+    }
+
+    if requester_pays:
+        headers['x-amz-request-payer'] = 'requester'
+    if public:
+        headers['x-amz-acl'] = 'public-read'
+    if os.environ.get('AWS_SESSION_TOKEN') and 'AWS_BUCKET_ACCESS_KEY_ID' not in os.environ:
+        headers['x-amz-security-token'] = os.environ.get('AWS_SESSION_TOKEN')
+    canonical_headers = '\n'.join('%s:%s' % (key, headers[key]) for key in sorted(headers)) + '\n'
+    signed_headers = ';'.join(sorted(headers.keys()))
+
+    canonical_request = '%s\n%s\n%s\n%s\n%s\n%s' % (
+        rtype, canonical_uri, canonical_querystring, canonical_headers, signed_headers, payload_hash
+    )
+    algorithm = 'AWS4-HMAC-SHA256'
+    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'
+    string_to_sign = algorithm + '\n' +  amzdate + '\n' +  credential_scope + '\n' +  hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
+    signing_key = getSignatureKey(secret_key, datestamp, region, service)
+    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
+    authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ', ' \
+        + 'SignedHeaders=' + signed_headers + ', ' + 'Signature=' + signature
+
+    request_url = 'https://%s%s' % (host, canonical_uri)
+    headers['Authorization'] = authorization_header
+    if content_type is not None:
+        headers['content-type'] = content_type
+    return request_url, headers

boto3utils/stepfunctions.py

@@ -0,0 +1,41 @@
+import boto3
+import json
+import logging
+
+from botocore.client import Config
+from botocore.vendored.requests.exceptions import ReadTimeout
+from traceback import format_exc
+
+logger = logging.getLogger(__name__)
+
+
+# module level client
+sfn = boto3.client('stepfunctions', config=Config(read_timeout=70))
+
+
+def run_activity(process, arn, **kwargs):
+    """ Run an activity around the process function provided """
+    while True:
+        logger.info('Querying for task')
+        try:
+            task = sfn.get_activity_task(activityArn=arn)
+        except ReadTimeout:
+            logger.warning('Activity read timed out')
+            continue
+        token = task.get('taskToken', None)
+        if token is None:
+            continue
+        logger.debug('taskToken: %s' % token)
+        try:
+            payload = task.get('input', '{}')
+            logger.info('Payload: %s' % payload)
+            # run process function with payload as kwargs
+            output = process(json.loads(payload))
+            # Send task success
+            sfn.send_task_success(taskToken=token, output=json.dumps(output))
+        except Exception as e:
+            err = str(e)
+            tb = format_exc()
+            logger.error("Exception when running task: %s - %s" % (err, json.dumps(tb)))
+            err = (err[252] + ' ...') if len(err) > 252 else err
+            sfn.send_task_failure(taskToken=token, error=str(err), cause=tb)

boto3utils/version.py

@@ -1 +1 @@
-__version__ = '0.0.2'
+__version__ = '0.0.3'

test/test_s3.py

@@ -2,6 +2,7 @@
 import os
 import pytest
 
+from datetime import datetime
 # this must be imported before any boto3 module
 from moto import mock_s3
 
@@ -77,4 +78,19 @@ def test_read_json(s3mock):
 def test_find(s3mock):
     urls = list(s3.find('s3://%s/test' % BUCKET))
     assert(len(urls) > 0)
-    assert('test.json' in urls)
+    assert('test.json' in urls)
+
+def test_latest_inventory():
+    url = 's3://sentinel-inventory/sentinel-s1-l1c/sentinel-s1-l1c-inventory'
+    suffix = 'productInfo.json'
+    for f in s3.latest_inventory(url, suffix=suffix):
+        dt = datetime.strptime(f['LastModifiedDate'], "%Y-%m-%dT%H:%M:%S.%fZ")
+        hours = (datetime.today() - dt).seconds // 3600
+        assert(hours < 24)
+        assert(f['url'].endswith(suffix))
+        break
+    for f in s3.latest_inventory(url):
+        dt = datetime.strptime(f['LastModifiedDate'], "%Y-%m-%dT%H:%M:%S.%fZ")
+        hours = (datetime.today() - dt).seconds // 3600
+        assert(hours < 24)
+        break

test/test_stepfunctions.py

@@ -0,0 +1,20 @@
+#import boto3
+#import os
+#import pytest
+
+# this must be imported before any boto3 module
+#from moto import mock_sfn
+
+#from boto3utils import stepfunctions as sfn
+#from shutil import rmtree
+
+
+#@pytest.fixture(scope='function')
+#def sfn_mock():
+#    with mock_sfn():
+#        client = boto3.client('stepfunctions')
+#        yield client
+
+
+#def test_run_activity(sfn):
+#    sfn.run_activity(sfn, arn)