Skip to main content

Utilities for building and testing AWS applications in Python

Project description

awstin

PyPI Dev Status

CI Build codecov

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

DynamoDB

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.

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.

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.

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.

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,
)

Float and Decimal

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

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.

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)

Future Work

See the Milestone for more information on what's currently planned for DynamoDB support. Feel free to open an issue for anything that's missing.

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.

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.

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:

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.

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:

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

awstin-0.0.13.tar.gz (20.8 kB view details)

Uploaded Source

Built Distribution

awstin-0.0.13-py3-none-any.whl (25.1 kB view details)

Uploaded Python 3

File details

Details for the file awstin-0.0.13.tar.gz.

File metadata

  • Download URL: awstin-0.0.13.tar.gz
  • Upload date:
  • Size: 20.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.3.0 pkginfo/1.6.1 requests/2.25.1 setuptools/49.2.1 requests-toolbelt/0.9.1 tqdm/4.55.0 CPython/3.9.1

File hashes

Hashes for awstin-0.0.13.tar.gz
Algorithm Hash digest
SHA256 e0e2d8aa1c4e30e1cfe0cc3a245b952c7ddb596ca6f6a378f3fd851727fc00ee
MD5 6a14ce09c865371b97ba69bde754974b
BLAKE2b-256 f161e464b19197fad2503325c508edf6755ef63243b62c9a096de97508fe8109

See more details on using hashes here.

File details

Details for the file awstin-0.0.13-py3-none-any.whl.

File metadata

  • Download URL: awstin-0.0.13-py3-none-any.whl
  • Upload date:
  • Size: 25.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.3.0 pkginfo/1.6.1 requests/2.25.1 setuptools/49.2.1 requests-toolbelt/0.9.1 tqdm/4.55.0 CPython/3.9.1

File hashes

Hashes for awstin-0.0.13-py3-none-any.whl
Algorithm Hash digest
SHA256 5fcbda13f311dd46080474f20fb670855ab2ae541fb3244f12647834917857e9
MD5 1feb0d4acbbe60bf133526bd03028941
BLAKE2b-256 be59957c6a43418a117f0d4250c7d15fbe2b8024ff43233a9462e0776ae18e82

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page