lambda-common ======== This library is designed to suppliment lambda's with commonly used connectors and utils. Installation ------------ ```bash pip install -i https://pypi.theorchard.io/pypi/ lambdacommon ``` Usage ----- #### Sql Connections This section talks about using SQL connection helpers. There are several helpers and several applicable ENV vars you can use. * `mysql_connection(host, user, password, database, connect_timeout=5, port=3306)`: * host (str): hostname of database server * user (str): user name * password (str): password * database (str): database name * connect_timeout (int): time to wait for db connection * port (int): the port of the sql connection * `dd_connection(connection_info=None)`: * If connection_info is left as None then it will look at `DD_DB_CREDENTIALS` which will be defined below * If you decide to pass data to connection_info it takes a dictionary with `mysql_connection` params * `ar_connection(connection_info=None)`: * If connection_info is left as None then it will look at `AR_DB_CREDENTIALS` which will be defined below * If you decide to pass data to connection_info it takes a dictionary with `mysql_connection` params * DD_DB_CREDENTIALS (common_config) * (env) DD_MYSQL_USER * (env) DD_MYSQL_PASSWORD * (env) DD_MYSQL_HOST * (env) DD_MYSQL_PORT (defaults to 3306) * (env) DD_MYSQL_DATABASE * AR_DB_CREDENTIALS (common_config) * (env) AR_MYSQL_USER * (env) AR_MYSQL_PASSWORD * (env) AR_MYSQL_HOST * (env) AR_MYSQL_PORT (defaults to 3306) * (env) AR_MYSQL_DATABASE Example: ```python from lambdacommon.util import ar_connection # Example of Art Relations Query SQL_QUERY = 'SELECT * FROM table' with ar_connection() as conn: with conn.cursor() as cursor: cursor.execute(SQL_QUERY) ``` You can also connect to multiple dbs if you want: ```python from lambdacommon.util import ar_connection from lambdacommon.util import dd_connection SQL_QUERY = 'SELECT * FROM table' with ar_connection() as ar_conn: with dd_connection() as dd_conn: with ar_conn.cursor() as cursor: cursor.execute(SQL_QUERY) with dd_conn.cursor() as cursor: cursor.execute(SQL_QUERY) ``` You can also set use them in pytest fixtures: ```python with util.dd_connection() as conn: with conn.cursor() as cursor: for query in teardown_queries: cursor.execute(query) for query in setup_queries: cursor.execute(query) conn.commit() yield for query in teardown_queries: cursor.execute(query) ``` #### Datadog This is a helper for sending data to datadog. https://datadogpy.readthedocs.io/en/latest/ * Required env vars * DATADOG_API_KEY * DATADOG_APP_KEY Example: ```python import time from lambdacommon.util import datadog_connection from lambdacommon.common_config import ENVIRONMENT data = [1,2,3] with datadog_connection() as datadog_conn: now = time.time() datadog_conn.Metric.send([ { 'metric': 'metric_namespace.metric_name', 'type': 'count', 'interval': 60, 'points': (now, len(data)), 'tags': ['environment:{}'.format(ENVIRONMENT)] }, { 'metric': 'metric_namespace.metric_name2', 'type': 'count', 'interval': 60, 'points': (now, len(data)), 'tags': ['environment:{}'.format(ENVIRONMENT)] } ]) ``` #### Logging This logging helper implements the non flask portion python-owslogger. https://github.com/theorchard/python-owslogger?ref=2.0.0 ```python from logging import INFO from owslogger import logger # Global logger app_logger = logger.setup( 'environment', 'logger_name', 'INFO', 'service_name', '1.0.0', dsn='loggly http/s url') # If you just want to set the logger in one time: current_app_logger = logger.setup( 'environment', 'logger_name', 'INFO', 'service_name', '1.0.0', correlation_id='correlation_id', dsn='loggly http/s url') ``` #### MSK Message Handling Lambdas that consume off of kafka topics receive an encoded list of N messages from the kafka topic. The [`LambdaMSKEventSourceMessage`](lambdacommon/models/lambda_msk_event_source_message.py) model provides a way to loop over each kafka message using a generator and the [`KafkaMessage`](lambdacommon/models/kafka_message.py) model to represent individual kafka messages. Example: ```python # inside lambda handler file def handler(event, context): my_message = LambdaMSKEventSourceMessage(event) for event_key, event in my_message: print(event.key, event.value) ``` #### Step Function Testing To simplify interacting with the AWS SDK for Step Functions, you can use the utility class: ```python # As part of your integration tests: from lambdacommon.aws.sfn import StateMachineExecutionTest def test_case_A(): # Set up a Test Run test_run_A = StateMachineExecutionTest( sfn_arn='arn:aws:states:us-east-1:1234567890:stateMachine:TestStateMachine', test_name='run_A', execution_input='{"key": "value"}' ) # Check if the same test run has been started in parallel if not test_run_A.was_started_parallel: # Start the State Machine test run if there are not parallel runs test_run_A.run() # Wait for the test run to finish test_run_A.wait_execution_end() # Assert it assert test_run_A.was_successful is True assert test_run_A.was_failed is False output = test_run_A.final_output assert output == 'Your Output' # Check certain tasks have been run (good to check ChoiceState paths) assert test_run_A.task_was_called('Your Task Name') is True # Get task info assert test_run_A.get_task('Your Task Name').state_type == 'ExpectedTaskType' ``` #### Lambda Invocation utils Trigger and test your Lambda functions for your integration tests. Example: ```python # In your integration tests code from lambdacommon.aws import lambdafunction def test_lambda(): test_lambda_A = lambdafunction.LambdaFunction('my-lambda-function-name') # Invoke your lambda synchronously, and assert it's response sync_test = test_lambda_A.invoke_synchronously( payload={'key': 'value'}, get_log=True ) assert sync_test['StatusCode'] == 200 response_payload = json.load(sync_test['Response']) assert response_payload == {'expected_key': 'expected_value'} # Invoke your lambda Asynchronously # This just sends the lambda into a queue to be invoked, nothing but a confirmation can get retrieved from this async_lambda = test_lambda_A.invoke_asynchronously(payload={'key': 'value'}) ```