Skip to main content

Provides a bunch of functions and utils to help you build applications on top of Amazon Web Services Resources

Project description

Fluxo AWS

Provides a bunch of functions and utils to help you build applications on top of Amazon Web Services Resources. We use this in our web development flow and would love hear suggestions and improvements on this.

Usage

Just import the function or class that you want to use, and you're good to go

Functions

AWS Lambda handlers

  1. prepare_response

Function Decorator that:

  • Allows you to make a response in a tuple format containing (status_code, body, headers). Example: return 200, {"success": True}
  • If body is dict or list, runs json.dumps function with encoders for datetime.datetime, decimal.Decimal and bytes
  • Add CORS headers
  • Add JSON content type header
  • Defaults your response: 200 for status code, {} for body

Usage:

from fluxo_aws import prepare_response

@prepare_response
def handler(event, context):
    return 200, {"success": True}
  1. event_parser

Function decorator that transform your event variable to a ParsedEvent class, exposing three methods: body, headers and query. It will transform strings into dict, and parse variables for you.

Usage

from fluxo_aws import prepare_response, event_parser

@prepare_response
@event_parser
def handler(event, context):
    print(event.body)
    print(event.headers)
    print(event.query)

    param = event.query.get("param")

    return 200

DynamoDB handlers

  1. DynamodbTable(table_name, schema, hash_key=None, partition_key=None)

Helper for DynamoDB. schema is a valid cerberus schema dict. This class exposes:

  • exists(id, hash_key=None): check if hash key exists in table, returning True of False
  • get_by_hash_key(id, hash_key=None): get a list of records for given hash key
  • add(data): insert dict into DynamoDB. Raise SchemaError if dict does not match schema with table schema

Usage

from fluxo_aws import DynamodbTable, SchemaError

schema = {"name": {"type": "string"}}

table = DynamodbTable("table", schema, hash_key=name)

try:
    table.add({"name": 123})
except SchemaError as e:
    print(f"schema error: {str(e)}")

print(table.exists("test"))
print(table.get_by_hash_key("test"))

Auth handlers

  1. hash_password(password)

Hashes and salt a password string using bcrypt and HS256

Usage

from fluxo_aws import hash_password

print(hash_password("secret"))
  1. verify_password(plain_password, hashed_password)

Compares a hash string with a password string, returning True if matches and False if not matches

Usage

from fluxo_aws import verify_password

print(verify_password("secret", "..."))
  1. create_access_token(data, expires_delta, secret_key)

Creates a JSON Web Token for data with an expiration delta of expires_delta

Usage

from fluxo_aws import create_access_token
from uuid import uuid4
from datetime import timedelta

print(create_access_token({"test": True}, timedelta(hours=3), str(uuid4())))
  1. decode_token(data, secret_key)

WIP

S3 handlers

  1. S3Bucket(bucket_name)

Helper for S3 Operations. This class exposes:

  • upload_file(file_name, object_name=None): upload local file to S3 returns True if uploaded successfully else False
  • download_file(object_name, file_name=None): download S3 file locally
  • create_presigned_url(object_name, action="get_object", expiration=3600): creates a presigned URL for S3 object. Returns presigned URL if successfully else returns None

Usage

from fluxo_aws import S3Bucket

s3_bucket = S3Bucket("my-test-bucket")
object_name = "test_file.txt"
file_name = "/tmp/text_file.txt"


print(s3_bucket.upload_file(file_name, object_name))

print(s3_bucket.create_presigned_url(object_name, action, expiration))

print(s3_bucket.download_file(object_name, file_name))

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

fluxo-aws-0.0.27.tar.gz (8.6 kB view details)

Uploaded Source

Built Distribution

fluxo_aws-0.0.27-py3-none-any.whl (9.2 kB view details)

Uploaded Python 3

File details

Details for the file fluxo-aws-0.0.27.tar.gz.

File metadata

  • Download URL: fluxo-aws-0.0.27.tar.gz
  • Upload date:
  • Size: 8.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.2.0 pkginfo/1.5.0.1 requests/2.24.0 setuptools/50.3.0 requests-toolbelt/0.9.1 tqdm/4.48.2 CPython/3.8.5

File hashes

Hashes for fluxo-aws-0.0.27.tar.gz
Algorithm Hash digest
SHA256 2d42d380c6711c9f3f6a39689bbbb2434a9ffb4d64cd89c439c04d0170bfd71e
MD5 aed2d9b043c7383670a1e5899077a0aa
BLAKE2b-256 5837d5c3e4f73429d602f2123ecee167ec29848de44497873a32cb2530623614

See more details on using hashes here.

File details

Details for the file fluxo_aws-0.0.27-py3-none-any.whl.

File metadata

  • Download URL: fluxo_aws-0.0.27-py3-none-any.whl
  • Upload date:
  • Size: 9.2 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.2.0 pkginfo/1.5.0.1 requests/2.24.0 setuptools/50.3.0 requests-toolbelt/0.9.1 tqdm/4.48.2 CPython/3.8.5

File hashes

Hashes for fluxo_aws-0.0.27-py3-none-any.whl
Algorithm Hash digest
SHA256 fa5ffe83a3bb1eddf755351a107151ce43aed28712e666bdf826cc4df06028d8
MD5 f35a0bbbb04f6a750c999f5984d9078a
BLAKE2b-256 569ee46ded5924c12fd0c709eee064cc75c78a2e5211097799860caf232e1079

See more details on using hashes here.

Supported by

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