Coverage for chalicelib/utils/storage/s3.py: 28%

1import boto3 

2import botocore 

3from botocore.client import Config 

4from botocore.exceptions import ClientError 

5from decouple import config 

6from requests.models import PreparedRequest 

7from chalicelib.utils.storage.interface import ObjectStorage 

8 

9 

10class AmazonS3Storage(ObjectStorage): 

11 if not config("S3_HOST", default=False): 11 ↛ 12line 11 didn't jump to line 12 because the condition on line 11 was never true

12 client = boto3.client('s3') 

13 resource = boto3.resource('s3') 

14 else: 

15 client = boto3.client('s3', endpoint_url=config("S3_HOST"), 

16 aws_access_key_id=config("S3_KEY"), 

17 aws_secret_access_key=config("S3_SECRET"), 

18 config=Config(signature_version='s3v4'), 

19 region_name=config("sessions_region"), 

20 verify=not config("S3_DISABLE_SSL_VERIFY", default=False, cast=bool)) 

21 resource = boto3.resource('s3', endpoint_url=config("S3_HOST"), 

22 aws_access_key_id=config("S3_KEY"), 

23 aws_secret_access_key=config("S3_SECRET"), 

24 config=Config(signature_version='s3v4'), 

25 region_name=config("sessions_region"), 

26 verify=not config("S3_DISABLE_SSL_VERIFY", default=False, cast=bool)) 

27 

28 def exists(self, bucket, key): 

29 try: 

30 self.resource.Object(bucket, key).load() 

31 except botocore.exceptions.ClientError as e: 

32 if e.response['Error']['Code'] == "404": 

33 return False 

34 else: 

35 # Something else has gone wrong. 

36 raise 

37 return True 

38 

39 def get_presigned_url_for_sharing(self, bucket, expires_in, key, check_exists=False): 

40 if check_exists and not self.exists(bucket, key): 

41 return None 

42 

43 return self.client.generate_presigned_url( 

44 'get_object', 

45 Params={ 

46 'Bucket': bucket, 

47 'Key': key 

48 }, 

49 ExpiresIn=expires_in 

50 ) 

51 

52 def get_presigned_url_for_upload(self, bucket, expires_in, key, **args): 

53 return self.client.generate_presigned_url( 

54 'put_object', 

55 Params={ 

56 'Bucket': bucket, 

57 'Key': key 

58 }, 

59 ExpiresIn=expires_in 

60 ) 

61 

62 def get_presigned_url_for_upload_secure(self, bucket, expires_in, key, conditions=None, public=False, 

63 content_type=None): 

64 acl = 'private' 

65 if public: 

66 acl = 'public-read' 

67 fields = {"acl": acl} 

68 if content_type: 

69 fields["Content-Type"] = content_type 

70 url_parts = self.client.generate_presigned_post( 

71 Bucket=bucket, 

72 Key=key, 

73 ExpiresIn=expires_in, 

74 Fields=fields, 

75 Conditions=conditions, 

76 ) 

77 req = PreparedRequest() 

78 req.prepare_url( 

79 f"{url_parts['url']}/{url_parts['fields']['key']}", url_parts['fields']) 

80 return req.url 

81 

82 def get_file(self, source_bucket, source_key): 

83 try: 

84 result = self.client.get_object( 

85 Bucket=source_bucket, 

86 Key=source_key 

87 ) 

88 except ClientError as ex: 

89 if ex.response['Error']['Code'] == 'NoSuchKey': 

90 return None 

91 else: 

92 raise ex 

93 return result["Body"].read().decode() 

94 

95 def tag_for_deletion(self, bucket, key): 

96 if not self.exists(bucket, key): 

97 return False 

98 # Copy the file to change the creation date, so it can be deleted X days after the tag's creation 

99 s3_target = self.resource.Object(bucket, key) 

100 s3_target.copy_from( 

101 CopySource={'Bucket': bucket, 'Key': key}, 

102 MetadataDirective='COPY', 

103 TaggingDirective='COPY' 

104 ) 

105 

106 self.tag_file(bucket=bucket, file_key=key, tag_key='to_delete_in_days', 

107 tag_value=config("SCH_DELETE_DAYS", default='7')) 

108 

109 def tag_file(self, file_key, bucket, tag_key, tag_value): 

110 return self.client.put_object_tagging( 

111 Bucket=bucket, 

112 Key=file_key, 

113 Tagging={ 

114 'TagSet': [ 

115 { 

116 'Key': tag_key, 

117 'Value': tag_value 

118 }, 

119 ] 

120 } 

121 )