# 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}
```
2. `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"))
```
2. `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", "..."))
```
3. `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())))
```
4. `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))
```