GeezerVison

aws_lambda/.DS_Store

Description

AWS Lambda is a service that allows you to write Python, Java, Node.js, C# or Go code that gets executed in response to events like http requests or files uploaded to S3.

AWS Lambda is a service that allows you to write Python, Java, or Node.js code that gets executed in response to events like http requests or files uploaded to S3.

Working with Lambda is relatively easy, but the process of bundling and deploying your code is not as simple as it could be.

import boto3

import botocore

import pip

import yaml

import subprocess

from .helpers import archive

from .helpers import get_environment_variable_value

from .helpers import mkdir

from .helpers import read

from .helpers import timestamp

from .helpers import LambdaContext

ARN_PREFIXES = {

# into a function we can execute.

fn = get_callable_handler_function(src, handler)

# TODO: look into mocking the ``context`` variable, currently being passed

# as None.

timeout = cfg.get('timeout')

if timeout:

context = LambdaContext(cfg.get('function_name'),timeout)

else:

context = LambdaContext(cfg.get('function_name'))

start = time.time()

results = fn(event, None)

results = fn(event, context)

end = time.time()

print('{0}'.format(results))

package = package.replace('-e ', '')

print('Installing {package}'.format(package=package))

pip_major_version = [int(v) for v in pip.__version__.split('.')][0]

if pip_major_version >= 10:

from pip._internal import main

main(['install', package, '-t', path, '--ignore-installed'])

else:

pip.main(['install', package, '-t', path, '--ignore-installed'])

subprocess.check_call([sys.executable, '-m', 'pip', 'install', package, '-t', path, '--ignore-installed'])

print ('Install directory contents are now: {directory}'.format(directory=os.listdir(path)))

def pip_install_to_target(path, requirements=None, local_package=None):

packages = []

if not requirements:

print('Gathering pip packages')

pip_major_version = [int(v) for v in pip.__version__.split('.')][0]

if pip_major_version >= 10:

from pip._internal import operations

packages.extend(operations.freeze.freeze())

else:

packages.extend(pip.operations.freeze.freeze())

pkgStr = subprocess.check_call([sys.executable, '-m', 'pip', 'freeze'])

packages.extend(pkgStr.decode('utf-8').splitlines())

else:

if os.path.exists(requirements):

print('Gathering requirement packages')

import datetime as dt

import os

import re

import time

import zipfile

if match is not None:

env_val = os.environ.get(match.group('environment_key_name'))

return env_val

class LambdaContext:

current_milli_time = lambda x: int(round(time.time() * 1000))

def get_remaining_time_in_millis(self):

return max(0, self.timeout_millis - (self.current_milli_time() - self.start_time_millis))

def __init__(self,function_name, timeoutSeconds = 3):

self.function_name = function_name

self.function_version = None

self.invoked_function_arn = None

self.memory_limit_in_mb = None

self.aws_request_id = None

self.log_group_name = None

self.log_stream_name = None

self.identity = None

self.client_context = None

self.timeout_millis = timeoutSeconds * 1000

self.start_time_millis = self.current_milli_time()

import sys

try: # for pip >= 10

from pip._internal.req import parse_requirements

except ImportError: # for pip <= 9.0.3

from pip.req import parse_requirements

try: # for pip >= 10

from pip._internal import download

except ImportError: # for pip <= 9.0.3

from pip import download

from setuptools import find_packages

from setuptools import setup

with open('README.rst') as readme_file:

readme = readme_file.read()

requirements = parse_requirements(

'requirements.txt', session=download.PipSession(),

pip_requirements = [str(r.req) for r in requirements]

requirements = [

'boto3',

]

if sys.version_info < (3, 0):

pip_requirements.append('futures')

requirements.append('futures')

test_requirements = [

# TODO: put package test requirements here

},

include_package_data=True,

scripts=['scripts/lambda'],

install_requires=pip_requirements,

install_requires=requirements,

license='ISCL',

zip_safe=False,

keywords='python-lambda',

from aws_lambda.helpers import LambdaContext

import time

import unittest

class TestLambdaContext(unittest.TestCase):

def test_get_remaining_time_in_millis(self):

context = LambdaContext('function_name',2000)

time.sleep(.5)

self.assertTrue(context.get_remaining_time_in_millis() < 2000)

if __name__ == '__main__':

unittest.main()