jshbrmn/irt-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add return types

Browse Source
This commit is contained in:
brmnjsh 2023-09-03 21:40:35 +00:00
parent d90f51faca
commit 0e05045817
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
app/helpers/aws_helper.py
View File

@ -16,21 +16,21 @@ else:
s3 = session.client('s3', region_name=ApplicationConfigs.region_name)
sqs = session.client('sqs', region_name=ApplicationConfigs.region_name)
def get_key_from_message(body: dict):
def get_key_from_message(body: dict) -> str:
return body['Records'][0]['s3']['object']['key']
def get_bucket_from_message(body: dict):
def get_bucket_from_message(body: dict) -> str:
return body['Records'][0]['s3']['bucket']['name']
def get_object(key: str, bucket: str):
def get_object(key: str, bucket: str) -> bytes:
return s3.get_object(
Bucket=bucket,
Key=key,
)['Body'].read()
def get_object_tag(key: str, bucket: str, tag_key: str):
def get_object_tag(key: str, bucket: str, tag_key: str) -> str | None:
tags = get_object_tags(key, bucket)
tag_index = 0
@ -41,7 +41,7 @@ def get_object_tag(key: str, bucket: str, tag_key: str):
else:
return None
def get_object_tags(key: str, bucket: str):
def get_object_tags(key: str, bucket: str) -> list:
tags = s3.get_object_tagging(Bucket=bucket, Key=key)['TagSet']
return tags