awstin


Nameawstin JSON
Version 0.0.12 PyPI version JSON
download
home_pagehttps://https://github.com/k2bd/awstin
SummaryUtilities for building and testing AWS applications in Python
upload_time2020-11-29 00:48:35
maintainer
docs_urlNone
authorKevin Duff
requires_python>=3.6
license
keywords
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # awstin

[![PyPI](https://img.shields.io/pypi/v/awstin)](https://pypi.org/project/awstin/) ![Dev Status](https://img.shields.io/pypi/status/awstin)

![CI Build](https://github.com/k2bd/awstin/workflows/CI/badge.svg) [![codecov](https://codecov.io/gh/k2bd/awstin/branch/master/graph/badge.svg)](https://codecov.io/gh/k2bd/awstin)

High-level utilities for building and testing AWS applications in Python.


## DynamoDB

### Production

The `DynamoDB` class currently gives dict-like access to boto3 `Table`s and their contents.
This requires either the `TEST_DYNAMODB_ENDPOINT` (for integration testing) or `AWS_REGION` (for production) environment variable to be set.

```python
from awstin.dynamodb import DynamoDB


dynamodb = DynamoDB()

# List of available tables
tables = dynamodb.list_tables()

# Access a table by name
table1 = dynamodb["my_table"]

# Tables that only have a partition key can be accessed directly by their
# partition key
item1 = table1["an_item"]

# Tables that have partition and sort keys can be accessed by a tuple
table2 = dynamodb["another_table"]
item2 = table2[("hashval", 123)]

# Full primary key access is also available
item3 = table2[{"hashkey_name": "hashval", "sortkey_name": 123}]
```

Tables can be scanned without worrying about pagination. `Table.scan` yields items, requesting another page of items lazily only when it's out of items in a page.


### Testing

For integration testing, a context manager to create and then automatically tear-down a DynamoDB table is provided.
The context manager waits for the table to be created/deleted before entering/exiting to avoid testing issues.
Hashkey and sortkey info can be provided.

```python
from awstin.dynamodb.testing import temporary_dynamodb_table


with temporary_dynamodb_table("table_name", "hashkey_name") as table:
    item = {
        "hashkey_name": "test_value",
        "another_key": 5,
    }
    table.put_item(item)
```


## Lambdas

### Production

Lambda handlers can be made more readable by separating event parsing from business logic.
The `lambda_handler` decorator factory takes a parser for the triggering event and context, and returns individual values to be used in the wrapped function.
```python
from awstin.awslambda import lambda_handler

def event_parser(event, context):
    request_id = event["requestContext"]["requestId"]
    memory_limit = context["memory_limit_in_mb"]
    return request_id, memory_limit


@lambda_handler(event_parser)
def handle_custom_event(request_id, memory_limit):
    print(request_id)
    print(memory_limit)
```


## API Gateway

### Authorization Lambdas

#### Production

Authorizor lambda responses can be generated with helper functions provided by `awstin.apigateway.auth`. `accept`, `reject`, `unauthorized`, and `invalid` will produce properly formatted auth lambda responses.

```python
from awstin.apigateway import auth


def auth_event_parser(event, _context):
    token = event["headers"]["AuthToken"]
    resource_arn = event["methodArn"]
    principal_id = event["requestContext"]["connectionId"]

    return token, resource_arn, principal_id


@lambda_handler(auth_event_parser)
def token_auth(token, resource_arn, principal_id):
    if token == "good token":
        return auth.accept(principal_id, resource_arn)
    elif token == "bad token":
        return auth.reject(principal_id, resource_arn)
    elif token == "unauthorized token":
        return auth.unauthorized()
    else:
        return auth.invalid()
```

### Websockets

#### Production

Websocket pushes can be performed with a callback URL and message:

```python
from awstin.apigateway.websocket import Websocket


Websocket("endpoint_url", "dev").send("callback_url", "message")
```


## SNS

### Production

SNS topics can be retrieved by name and published to with the message directly.
This requires either the `TEST_SNS_ENDPOINT` (for integration testing) or `AWS_REGION` (for production) environment variable to be set.

```python
from awstin.sns import SNSTopic


topic = SNSTopic("topic-name")
message_id = topic.publish("a message")
```

Message attributes can be set from the kwargs of the publish:

```python
topic.publish(
    "another message",
    attrib_a="a string",
    attrib_b=1234,
    attrib_c=["a", "b", False, None],
    attrib_d=b"bytes value",
)
```



            

Raw data

            {
    "_id": null,
    "home_page": "https://https://github.com/k2bd/awstin",
    "name": "awstin",
    "maintainer": "",
    "docs_url": null,
    "requires_python": ">=3.6",
    "maintainer_email": "",
    "keywords": "",
    "author": "Kevin Duff",
    "author_email": "",
    "download_url": "https://files.pythonhosted.org/packages/63/cd/1bb9c218e0f389bdd84daa25584a9b07de2a1977e8b868f2b8c41cdd638e/awstin-0.0.12.tar.gz",
    "platform": "",
    "description": "# awstin\n\n[![PyPI](https://img.shields.io/pypi/v/awstin)](https://pypi.org/project/awstin/) ![Dev Status](https://img.shields.io/pypi/status/awstin)\n\n![CI Build](https://github.com/k2bd/awstin/workflows/CI/badge.svg) [![codecov](https://codecov.io/gh/k2bd/awstin/branch/master/graph/badge.svg)](https://codecov.io/gh/k2bd/awstin)\n\nHigh-level utilities for building and testing AWS applications in Python.\n\n\n## DynamoDB\n\n### Production\n\nThe `DynamoDB` class currently gives dict-like access to boto3 `Table`s and their contents.\nThis requires either the `TEST_DYNAMODB_ENDPOINT` (for integration testing) or `AWS_REGION` (for production) environment variable to be set.\n\n```python\nfrom awstin.dynamodb import DynamoDB\n\n\ndynamodb = DynamoDB()\n\n# List of available tables\ntables = dynamodb.list_tables()\n\n# Access a table by name\ntable1 = dynamodb[\"my_table\"]\n\n# Tables that only have a partition key can be accessed directly by their\n# partition key\nitem1 = table1[\"an_item\"]\n\n# Tables that have partition and sort keys can be accessed by a tuple\ntable2 = dynamodb[\"another_table\"]\nitem2 = table2[(\"hashval\", 123)]\n\n# Full primary key access is also available\nitem3 = table2[{\"hashkey_name\": \"hashval\", \"sortkey_name\": 123}]\n```\n\nTables can be scanned without worrying about pagination. `Table.scan` yields items, requesting another page of items lazily only when it's out of items in a page.\n\n\n### Testing\n\nFor integration testing, a context manager to create and then automatically tear-down a DynamoDB table is provided.\nThe context manager waits for the table to be created/deleted before entering/exiting to avoid testing issues.\nHashkey and sortkey info can be provided.\n\n```python\nfrom awstin.dynamodb.testing import temporary_dynamodb_table\n\n\nwith temporary_dynamodb_table(\"table_name\", \"hashkey_name\") as table:\n    item = {\n        \"hashkey_name\": \"test_value\",\n        \"another_key\": 5,\n    }\n    table.put_item(item)\n```\n\n\n## Lambdas\n\n### Production\n\nLambda handlers can be made more readable by separating event parsing from business logic.\nThe `lambda_handler` decorator factory takes a parser for the triggering event and context, and returns individual values to be used in the wrapped function.\n```python\nfrom awstin.awslambda import lambda_handler\n\ndef event_parser(event, context):\n    request_id = event[\"requestContext\"][\"requestId\"]\n    memory_limit = context[\"memory_limit_in_mb\"]\n    return request_id, memory_limit\n\n\n@lambda_handler(event_parser)\ndef handle_custom_event(request_id, memory_limit):\n    print(request_id)\n    print(memory_limit)\n```\n\n\n## API Gateway\n\n### Authorization Lambdas\n\n#### Production\n\nAuthorizor lambda responses can be generated with helper functions provided by `awstin.apigateway.auth`. `accept`, `reject`, `unauthorized`, and `invalid` will produce properly formatted auth lambda responses.\n\n```python\nfrom awstin.apigateway import auth\n\n\ndef auth_event_parser(event, _context):\n    token = event[\"headers\"][\"AuthToken\"]\n    resource_arn = event[\"methodArn\"]\n    principal_id = event[\"requestContext\"][\"connectionId\"]\n\n    return token, resource_arn, principal_id\n\n\n@lambda_handler(auth_event_parser)\ndef token_auth(token, resource_arn, principal_id):\n    if token == \"good token\":\n        return auth.accept(principal_id, resource_arn)\n    elif token == \"bad token\":\n        return auth.reject(principal_id, resource_arn)\n    elif token == \"unauthorized token\":\n        return auth.unauthorized()\n    else:\n        return auth.invalid()\n```\n\n### Websockets\n\n#### Production\n\nWebsocket pushes can be performed with a callback URL and message:\n\n```python\nfrom awstin.apigateway.websocket import Websocket\n\n\nWebsocket(\"endpoint_url\", \"dev\").send(\"callback_url\", \"message\")\n```\n\n\n## SNS\n\n### Production\n\nSNS topics can be retrieved by name and published to with the message directly.\nThis requires either the `TEST_SNS_ENDPOINT` (for integration testing) or `AWS_REGION` (for production) environment variable to be set.\n\n```python\nfrom awstin.sns import SNSTopic\n\n\ntopic = SNSTopic(\"topic-name\")\nmessage_id = topic.publish(\"a message\")\n```\n\nMessage attributes can be set from the kwargs of the publish:\n\n```python\ntopic.publish(\n    \"another message\",\n    attrib_a=\"a string\",\n    attrib_b=1234,\n    attrib_c=[\"a\", \"b\", False, None],\n    attrib_d=b\"bytes value\",\n)\n```\n\n\n",
    "bugtrack_url": null,
    "license": "",
    "summary": "Utilities for building and testing AWS applications in Python",
    "version": "0.0.12",
    "split_keywords": [],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "md5": "3eb6036f6b80776c8dc188b0807fc2fe",
                "sha256": "626a21d9aeb041df5521209a7e03698381b44e3023392ef09ce70365d950a46e"
            },
            "downloads": -1,
            "filename": "awstin-0.0.12-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "3eb6036f6b80776c8dc188b0807fc2fe",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.6",
            "size": 19510,
            "upload_time": "2020-11-29T00:48:34",
            "upload_time_iso_8601": "2020-11-29T00:48:34.362297Z",
            "url": "https://files.pythonhosted.org/packages/70/8a/965a041a88d051d3290780c2e9037dd00f54e0de93b937a971f718134b39/awstin-0.0.12-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": "",
            "digests": {
                "md5": "ca33e68d24cc48ccd8173aa8172da1d9",
                "sha256": "31b71c2250584235f3f6b1d04f9ba7297aba6564e9849c804be9dcc461b4c860"
            },
            "downloads": -1,
            "filename": "awstin-0.0.12.tar.gz",
            "has_sig": false,
            "md5_digest": "ca33e68d24cc48ccd8173aa8172da1d9",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.6",
            "size": 14557,
            "upload_time": "2020-11-29T00:48:35",
            "upload_time_iso_8601": "2020-11-29T00:48:35.266301Z",
            "url": "https://files.pythonhosted.org/packages/63/cd/1bb9c218e0f389bdd84daa25584a9b07de2a1977e8b868f2b8c41cdd638e/awstin-0.0.12.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2020-11-29 00:48:35",
    "github": false,
    "gitlab": false,
    "bitbucket": false,
    "lcname": "awstin"
}
        
Elapsed time: 0.23677s