Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/usr/bin/env python
- import argparse
- import zipfile
- import boto3
- import tempfile
- import os
- import yaml
- import json
- import stat
- import logging
- import sys
- BASE_DIR = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'))
- def fetch_file_path(dpath, root_dir):
- tf = []
- for root, dirs, files in os.walk(dpath):
- for fname in files:
- if not root.endswith('/__pycache__'):
- fpath = os.path.join(root, fname)
- tf.append((fpath, fpath[len(root_dir) + 1:]))
- return tf
- def pack_zip_file(config):
- fd1, tpath = tempfile.mkstemp('.zip')
- os.close(fd1)
- target_files = []
- pkg_dir = os.path.normpath(
- os.path.join(os.path.dirname(boto3.__path__[0]), '..')
- )
- target_files = fetch_file_path(pkg_dir, pkg_dir) + \
- fetch_file_path(os.path.join(BASE_DIR, 'bin'), BASE_DIR) + \
- fetch_file_path(os.path.join(BASE_DIR, 'slam'), BASE_DIR)
- with zipfile.ZipFile(tpath, 'w', zipfile.ZIP_DEFLATED) as z:
- for fpath, wpath in target_files:
- if wpath.startswith('boto3-') or wpath.startswith('botocore-'):
- continue
- z.write(fpath, wpath)
- fd2, jpath = tempfile.mkstemp('.json')
- os.write(fd2, json.dumps(config).encode('utf8'))
- os.close(fd2)
- os.chmod(jpath, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
- z.write(jpath, 'config.json')
- return tpath
- def configure_lambda(client, raw_data, config):
- logger.info('configure {} -> {}'.format(config['script-name'],
- config['function-name']))
- code_res = client.update_function_code(
- FunctionName=config['function-name'],
- ZipFile=raw_data,
- )
- logger.debug(' Upload code: %s', json.dumps(code_res, indent=4))
- conf_res = client.update_function_configuration(
- FunctionName=config['function-name'],
- Handler='bin.{}.lambda_handler'.format(config['script-name']),
- Timeout=config.get('timeout', 60),
- MemorySize=config.get('memory', 128),
- Runtime='python3.6'
- )
- logger.debug(' Update config: %s', json.dumps(conf_res, indent=4))
- def main():
- psr = argparse.ArgumentParser()
- psr.add_argument('-c', '--config',
- default=os.path.join(BASE_DIR, 'config.yml'), type=open)
- args = psr.parse_args()
- config = yaml.load(args.config)
- zip_fpath = pack_zip_file(config)
- client = boto3.client('lambda')
- raw_data = open(zip_fpath, 'rb').read()
- for func in config['functions']:
- configure_lambda(client, raw_data, func)
- if __name__ == '__main__':
- logger = logging.getLogger()
- logger.setLevel(logging.INFO)
- logger.addHandler(logging.StreamHandler(sys.stdout))
- main()
Add Comment
Please, Sign In to add comment