2022-07-07 13:28:49 +09:00

59 lines
2.0 KiB
Python

import boto3
import environments
from constants import AWS_RESOURCE_S3, S3_RESPONSE_BODY, UTF8
class S3Resource:
def __init__(self, bucket_name: str) -> None:
self.__s3_resource = boto3.resource(AWS_RESOURCE_S3)
self.__s3_bucket = self.__s3_resource.Bucket(bucket_name)
def get_object(self, object_key: str):
s3_object = self.__s3_bucket.Object(object_key)
response = s3_object.get()
return response[S3_RESPONSE_BODY].read().decode(UTF8)
class ConfigBucket:
__s3_resource: S3Resource = None
__bucket_name: str
__check_target_schema_names_file_path: str
__notice_mail_title_template_file_path: str
__notice_mail_body_template_file_path: str
def __init__(self) -> None:
self.__bucket_name = environments.CONFIG_BUCKET_NAME
self.__check_target_schema_names_file_path = environments.CHECK_TARGET_SCHEMA_NAMES_PATH
self.__notice_mail_title_template_file_path = environments.NOTICE_MAIL_TITLE_TEMPLATE_PATH
self.__notice_mail_body_template_file_path = environments.NOTICE_MAIL_BODY_TEMPLATE_PATH
self.__s3_resource = S3Resource(self.__bucket_name)
@property
def bucket_name(self):
return self.__bucket_name
@property
def check_target_schema_names_file_path(self):
return self.__check_target_schema_names_file_path
@property
def mail_body_file_path(self):
return self.__notice_mail_body_template_file_path
@property
def mail_title_file_path(self):
return self.__notice_mail_title_template_file_path
@property
def check_target_schema_names(self):
return self.__s3_resource.get_object(self.__check_target_schema_names_file_path)
@property
def notice_mail_title_template(self):
return self.__s3_resource.get_object(self.__notice_mail_title_template_file_path)
@property
def notice_mail_body_template(self):
return self.__s3_resource.get_object(self.__notice_mail_body_template_file_path)