awstin


Nameawstin JSON
Version 0.0.22 PyPI version JSON
download
home_pagehttps://https://github.com/k2bd/awstin
SummaryUtilities for building and testing AWS applications in Python
upload_time2021-04-14 15:25:10
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)
[![Documentation Status](https://readthedocs.org/projects/awstin/badge/?version=latest)](https://awstin.readthedocs.io/en/latest/?badge=latest)
[![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

[![DynamoDB](https://img.shields.io/github/milestones/progress/k2bd/awstin/1)](https://github.com/k2bd/awstin/milestone/1)

### Production

To use DynamoDB either the `TEST_DYNAMODB_ENDPOINT` (for integration
testing) or `AWS_REGION` (for production) environment variable must be set.

DynamoDB is accessed through Python data models that users define to represent
structured data in tables.

```python
from awstin.dynamodb import Attr, DynamoModel, Key


class User(DynamoModel):
    # Name of the DynamoDB table (required!)
    _table_name_ = "Users"

    # Sort or hash keys are marked with Key
    user_id = Key()

    # Other attributes are marked with Attr
    favorite_color = Attr()

    # The names of attributes and keys can differ from the names on the data
    # model - the name of the attribute in DynamoDB should be passed to Attr
    account_age = Attr("ageDays")
```

Tables are tied to these data models. They'll be returned when items are 
retrieved from the table. Also, `put_item` takes instances of this data model class.

These data models also define projection expressions, so only those attributes
are retrieved from `get_item`, `query`, and `scan` calls.

```python
from awstin.dynamodb import DynamoDB


dynamodb = DynamoDB()

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

# Access a table by model
users_table = dynamodb[User]

# Put an item into the table
user = User(
    user_id="user123",
    favorite_color="Blue",
    account_age=120,
)
users_table.put_item(user)

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

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

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

Query and scan filters can be built up using these data models as well. Results can be iterated without worrying about pagination. `Table.scan` and `Table.query` yield items, requesting another page of items lazily only when it's out of items in a page.

```python
scan_filter = (
    (User.account_age > 30)
    & (User.favorite_color.in_(["Blue", "Green"]))
)

for user in users_table.scan(scan_filter):
    ban_user(user)
```

Queries must be given a query expression and can optionally be given a filter expression. Query expressions must represent valid DynamoDB queries.

```python
class Student(DynamoModel):
    _table_name_ = "Students"

    # Hash key
    name = Key()

    # Sort key
    year = Key()

    homeroom = Attr()


students_table = dynamodb[Student]

query_expression = (Student.name == "John") & (Student.year >= 10)
filter_expression = Student.homeroom == "Smith"

results = students_table.query(
    query_expression=query_expression,
    filter_expression=filter_expression,
)
```

Indexes work identically, but must have a `_index_name_` attribute on the data
model. Indexes can be used for queries and scans.

```python
class ByHomeroomIndex(DynamoModel):
    _table_name_ = "Students"
    _index_name_ = "ByHomeroom"

    # Hash key
    homeroom = Key()

    # Sort key
    name = Key()

    year = Attr()


homeroom_index = dynamodb[ByHomeroomIndex]

query_expression = (
    (ByHomeroomIndex.homeroom == "Doe")
    & (ByHomeroomIndex.name > "B")
)
filter_expression = ByHomeroomIndex.year > 11

items = list(homeroom_index.query(query_expression, filter_expression))
```

**Nested Values**

Filters on nested attributes work as well:

```python
scan_filter = (
    (MyModel.map_attr.key == "value")
    & (MyModel.list_attr[3] == 10)
)

results = my_table.scan(scan_filter)
```

**Updating Items**

A syntax is also available for updating items, with an optional condition expression:

```python
update_expression = (
    MyModel.an_attr.set(5 - MyModel.another_attr)
    & MyModel.third_attr.add(100)
    & MyModel.another_attr.remove()
    & MyModel.set_attr.delete([2, 3])
)

condition_expression = MyModel.an_attr > 11

updated = my_table.update_item(
    "primary_key",
    update_expression,
    condition_expression,
)
```

`if_not_exists` and `list_append` are provided as well:

```python
from awstin.dynamodb import list_append

update_expression = (
    MyModel.an_attr.set(MyModel.an_attr.if_not_exists(MyModel.another_attr))
    & MyModel.third_attr.set(list_append([1.1, 2.2], MyModel.list_attr))
)
```

`update_item` returns `None` if the condition evaluates to `False`.

**Float and Decimal**

Floats should be used when working with DynamoDB through `awstin`. Conversions between float and Decimal is done internally.


**Unset Values**

Values in a data model class that are unset, either by user instantiation or by
retrieval from DynamoDB, are given the value `awstin.dynamodb.NOT_SET`.

### 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(User, "hashkey_name") as table:
    item = User(
        user_id="user456",
        favorite_color="Green",
        account_age=333,
    )
    table.put_item(item)
```


## Lambdas

[![Lambda](https://img.shields.io/github/milestones/progress/k2bd/awstin/3)]((https://github.com/k2bd/awstin/milestone/3))

### 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)
```

#### Testing

A function wrapped with `lambda_handler` is stored on the `inner` attribute of the returned function. That way, the business logic of the handler can be tested separately without having to build events.

```python
@lambda_handler(my_parser)
def my_handler(a: int, b: str):
    ...

# ------

def test_parser():
    args = my_parser(test_event, test_context)
    assert ...

def test_handler():
    result = my_handler.inner(1, "abc")
    assert ...
```


## 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

[![SNS](https://img.shields.io/github/milestones/progress/k2bd/awstin/2)]((https://github.com/k2bd/awstin/milestone/2))

### 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/f9/41/2fe4756fdd6f09de5b1a4320e1b5b78e524e5f134c59592fa7305058f27d/awstin-0.0.22.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)\n[![Documentation Status](https://readthedocs.org/projects/awstin/badge/?version=latest)](https://awstin.readthedocs.io/en/latest/?badge=latest)\n[![codecov](https://codecov.io/gh/k2bd/awstin/branch/master/graph/badge.svg)](https://codecov.io/gh/k2bd/awstin)\n\n\nHigh-level utilities for building and testing AWS applications in Python.\n\n\n## DynamoDB\n\n[![DynamoDB](https://img.shields.io/github/milestones/progress/k2bd/awstin/1)](https://github.com/k2bd/awstin/milestone/1)\n\n### Production\n\nTo use DynamoDB either the `TEST_DYNAMODB_ENDPOINT` (for integration\ntesting) or `AWS_REGION` (for production) environment variable must be set.\n\nDynamoDB is accessed through Python data models that users define to represent\nstructured data in tables.\n\n```python\nfrom awstin.dynamodb import Attr, DynamoModel, Key\n\n\nclass User(DynamoModel):\n    # Name of the DynamoDB table (required!)\n    _table_name_ = \"Users\"\n\n    # Sort or hash keys are marked with Key\n    user_id = Key()\n\n    # Other attributes are marked with Attr\n    favorite_color = Attr()\n\n    # The names of attributes and keys can differ from the names on the data\n    # model - the name of the attribute in DynamoDB should be passed to Attr\n    account_age = Attr(\"ageDays\")\n```\n\nTables are tied to these data models. They'll be returned when items are \nretrieved from the table. Also, `put_item` takes instances of this data model class.\n\nThese data models also define projection expressions, so only those attributes\nare retrieved from `get_item`, `query`, and `scan` calls.\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 model\nusers_table = dynamodb[User]\n\n# Put an item into the table\nuser = User(\n    user_id=\"user123\",\n    favorite_color=\"Blue\",\n    account_age=120,\n)\nusers_table.put_item(user)\n\n# Tables that only have a partition key can be accessed directly by their\n# partition key\nitem1 = users_table[\"user123\"]\n\n# Tables that have partition and sort keys can be accessed by a tuple\ntable2 = dynamodb[AnotherTableModel]\nitem2 = table2[(\"hashval\", 123)]\n\n# Full primary key access is also available\nitem3 = table2[{\"hashkey_name\": \"hashval\", \"sortkey_name\": 123}]\n```\n\nQuery and scan filters can be built up using these data models as well. Results can be iterated without worrying about pagination. `Table.scan` and `Table.query` yield items, requesting another page of items lazily only when it's out of items in a page.\n\n```python\nscan_filter = (\n    (User.account_age > 30)\n    & (User.favorite_color.in_([\"Blue\", \"Green\"]))\n)\n\nfor user in users_table.scan(scan_filter):\n    ban_user(user)\n```\n\nQueries must be given a query expression and can optionally be given a filter expression. Query expressions must represent valid DynamoDB queries.\n\n```python\nclass Student(DynamoModel):\n    _table_name_ = \"Students\"\n\n    # Hash key\n    name = Key()\n\n    # Sort key\n    year = Key()\n\n    homeroom = Attr()\n\n\nstudents_table = dynamodb[Student]\n\nquery_expression = (Student.name == \"John\") & (Student.year >= 10)\nfilter_expression = Student.homeroom == \"Smith\"\n\nresults = students_table.query(\n    query_expression=query_expression,\n    filter_expression=filter_expression,\n)\n```\n\nIndexes work identically, but must have a `_index_name_` attribute on the data\nmodel. Indexes can be used for queries and scans.\n\n```python\nclass ByHomeroomIndex(DynamoModel):\n    _table_name_ = \"Students\"\n    _index_name_ = \"ByHomeroom\"\n\n    # Hash key\n    homeroom = Key()\n\n    # Sort key\n    name = Key()\n\n    year = Attr()\n\n\nhomeroom_index = dynamodb[ByHomeroomIndex]\n\nquery_expression = (\n    (ByHomeroomIndex.homeroom == \"Doe\")\n    & (ByHomeroomIndex.name > \"B\")\n)\nfilter_expression = ByHomeroomIndex.year > 11\n\nitems = list(homeroom_index.query(query_expression, filter_expression))\n```\n\n**Nested Values**\n\nFilters on nested attributes work as well:\n\n```python\nscan_filter = (\n    (MyModel.map_attr.key == \"value\")\n    & (MyModel.list_attr[3] == 10)\n)\n\nresults = my_table.scan(scan_filter)\n```\n\n**Updating Items**\n\nA syntax is also available for updating items, with an optional condition expression:\n\n```python\nupdate_expression = (\n    MyModel.an_attr.set(5 - MyModel.another_attr)\n    & MyModel.third_attr.add(100)\n    & MyModel.another_attr.remove()\n    & MyModel.set_attr.delete([2, 3])\n)\n\ncondition_expression = MyModel.an_attr > 11\n\nupdated = my_table.update_item(\n    \"primary_key\",\n    update_expression,\n    condition_expression,\n)\n```\n\n`if_not_exists` and `list_append` are provided as well:\n\n```python\nfrom awstin.dynamodb import list_append\n\nupdate_expression = (\n    MyModel.an_attr.set(MyModel.an_attr.if_not_exists(MyModel.another_attr))\n    & MyModel.third_attr.set(list_append([1.1, 2.2], MyModel.list_attr))\n)\n```\n\n`update_item` returns `None` if the condition evaluates to `False`.\n\n**Float and Decimal**\n\nFloats should be used when working with DynamoDB through `awstin`. Conversions between float and Decimal is done internally.\n\n\n**Unset Values**\n\nValues in a data model class that are unset, either by user instantiation or by\nretrieval from DynamoDB, are given the value `awstin.dynamodb.NOT_SET`.\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(User, \"hashkey_name\") as table:\n    item = User(\n        user_id=\"user456\",\n        favorite_color=\"Green\",\n        account_age=333,\n    )\n    table.put_item(item)\n```\n\n\n## Lambdas\n\n[![Lambda](https://img.shields.io/github/milestones/progress/k2bd/awstin/3)]((https://github.com/k2bd/awstin/milestone/3))\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#### Testing\n\nA function wrapped with `lambda_handler` is stored on the `inner` attribute of the returned function. That way, the business logic of the handler can be tested separately without having to build events.\n\n```python\n@lambda_handler(my_parser)\ndef my_handler(a: int, b: str):\n    ...\n\n# ------\n\ndef test_parser():\n    args = my_parser(test_event, test_context)\n    assert ...\n\ndef test_handler():\n    result = my_handler.inner(1, \"abc\")\n    assert ...\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[![SNS](https://img.shields.io/github/milestones/progress/k2bd/awstin/2)]((https://github.com/k2bd/awstin/milestone/2))\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.22",
    "split_keywords": [],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "md5": "ea7528dff63556e3b7cf6667a7b035c8",
                "sha256": "b2b704e7784834f78dfc2e65d8c0d25198f6148089edede332813224a4062853"
            },
            "downloads": -1,
            "filename": "awstin-0.0.22-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "ea7528dff63556e3b7cf6667a7b035c8",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.6",
            "size": 31999,
            "upload_time": "2021-04-14T15:25:08",
            "upload_time_iso_8601": "2021-04-14T15:25:08.938680Z",
            "url": "https://files.pythonhosted.org/packages/59/89/40aa807fcf8fa53af8756dcfcaff1aa81859372fc11f9abad259f5bdbfd6/awstin-0.0.22-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": "",
            "digests": {
                "md5": "b29b53044668e79198f4ca1dee985099",
                "sha256": "471dd8de1b99e9c725bdaceed8d6f9709a55bb7b3e7d375f781ec163adfc1d69"
            },
            "downloads": -1,
            "filename": "awstin-0.0.22.tar.gz",
            "has_sig": false,
            "md5_digest": "b29b53044668e79198f4ca1dee985099",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.6",
            "size": 28481,
            "upload_time": "2021-04-14T15:25:10",
            "upload_time_iso_8601": "2021-04-14T15:25:10.160393Z",
            "url": "https://files.pythonhosted.org/packages/f9/41/2fe4756fdd6f09de5b1a4320e1b5b78e524e5f134c59592fa7305058f27d/awstin-0.0.22.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2021-04-14 15:25:10",
    "github": false,
    "gitlab": false,
    "bitbucket": false,
    "lcname": "awstin"
}
        
Elapsed time: 0.30969s