Adapters following hexagonal architecture to connect various AWS services.
Project description
aws-hexagonal-adapters
Adapters following hexagonal architecture to connect various AWS services
Example usage
S3
Download object
from os import environ
from aws_hexagonal_adapters.s3_service import S3Service
S3_SERVICE = S3Service(region_name=environ["CURRENT_REGION"])
# Download file
S3_SERVICE.download(
bucket="bucket_name", local_path=f"/tmp/object", remote_path="object"
)
Upload object
from os import environ
from aws_hexagonal_adapters.s3_service import S3Service
S3_SERVICE = S3Service(region_name=environ["CURRENT_REGION"])
# Download file
S3_SERVICE.upload(
bucket="bucket_name",
local_path=f"/tmp/object_name",
remote_path="object_name",
extra_args={"StorageClass": "STANDARD_IA"},
)
List objects
from os import environ
from aws_hexagonal_adapters.s3_service import S3Service
S3_SERVICE = S3Service(region_name=environ["CURRENT_REGION"])
# Download file
S3_SERVICE.list_files(bucket="bucket_name", prefix="folder_name")
Delete object
from os import environ
from aws_hexagonal_adapters.s3_service import S3Service
S3_SERVICE = S3Service(region_name=environ["CURRENT_REGION"])
# Download file
S3_SERVICE.delete_object(bucket="bucket_name", key="object_name")
Delete objects
from os import environ
from aws_hexagonal_adapters.s3_service import S3Service
S3_SERVICE = S3Service(region_name=environ["CURRENT_REGION"])
objects = ["folder/object1.txt", "folder/object2.txt", "folder/object3.txt"]
# Download file
S3_SERVICE.delete_objects(bucket="bucket_name", keys=objects)
DynamoDB
Get Item
from os import environ
from aws_hexagonal_adapters.dynamo_db_service import DynamoDBService
DYNAMODB_SERVICE = DynamoDBService(region_name=environ["REGION_NAME"])
item = DYNAMODB_SERVICE.get_item(
table_name=environ["DYNAMODB_TABLE_NAME"],
key={"pk": f'{environ["DYNAMODB_TABLE_ITEM_NAME"]}', "sk": "1"},
)
Get Items
from os import environ
from boto3.dynamodb.conditions import Key
from aws_hexagonal_adapters.dynamo_db_service import DynamoDBService
DYNAMODB_SERVICE = DynamoDBService(region_name=environ["REGION_NAME"])
item = DYNAMODB_SERVICE.get_items(
table_name=environ["DYNAMODB_TABLE_NAME"],
filter_expression=Key("sk").eq("2"),
)
Scan
from os import environ
from aws_hexagonal_adapters.dynamo_db_service import DynamoDBService
DYNAMODB_SERVICE = DynamoDBService(region_name=environ["REGION_NAME"])
item = DYNAMODB_SERVICE.scan_items(
table_name=environ["DYNAMODB_TABLE_NAME"],
index_name=environ["DYNAMODB_INDEX_NAME"],
)
SQS
from os import environ
from aws_hexagonal_adapters.sqs_service import SQSService
SQS_SERVICE = SQSService(region_name=environ["REGION_NAME"])
messages = SQS_SERVICE.receive_messages(queue_url=environ["SQS_QUEUE_NAME"])
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for aws_hexagonal_adapters-1.2.2.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | ba490bebb7981cd0ae42e41074d108c8d18d2e06839c6429a7ab8b52801a8b54 |
|
MD5 | 45c9f5d72ac877621670d354287ca400 |
|
BLAKE2b-256 | fd6f42cec7f8b41ebe5c140c9f2050901eeaa87b4a338677e66a173c2bba95fc |
Close
Hashes for aws_hexagonal_adapters-1.2.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 0f31b83899b2680c1613937275d22dd07427f7dc53a9e0a89b4cc9ec43800cb0 |
|
MD5 | 61f889dd9ddb50e63d4d7fe249bd25bb |
|
BLAKE2b-256 | f1a66f097cc16137c568c228e4ec862f3ee9330b1f34d3aac86bd42eeb55e470 |