Coverage for website/thaliawebsite/storage/backend.py: 66.00%

1import os 

2 

3from django.conf import settings 

4from django.core import signing 

5from django.core.files.storage import FileSystemStorage 

6 

7from storages.backends.s3boto3 import S3Boto3Storage, S3ManifestStaticStorage 

8 

9 

10class S3RenameMixin: 

11 def rename(self, old_name, new_name): 

12 self.bucket.Object(new_name).copy_from( 

13 CopySource={"Bucket": self.bucket.name, "Key": old_name} 

14 ) 

15 self.delete(old_name) 

16 

17 

18class PublicS3Storage(S3RenameMixin, S3Boto3Storage): 

19 """Storage on S3 for public media files. 

20 

21 This storage is used for files that are publicly accessible. We use CloudFront to 

22 serve files from S3. CloudFront is configured with a behavior to serve files in the 

23 PUBLIC_MEDIA_LOCATION publicly, despite the objects in the bucket having a "private" 

24 ACL. Hence, we can still use the default "private" ACL to prevent people from using 

25 the S3 bucket directly. 

26 """ 

27 

28 location = settings.PUBLIC_MEDIA_LOCATION 

29 file_overwrite = False 

30 querystring_auth = False 

31 

32 # Cloudfront will have a behavior to serve files in PUBLIC_MEDIA_LOCATION publicly, 

33 # despite the objects in the bucket having a private ACL. Hence, we can use the 

34 # default "private" ACL to prevent people from reading from the bucket directly. 

35 

36 def url(self, name, attachment=False, expire_seconds=None): 

37 params = {} 

38 if attachment: 

39 # The S3 bucket will add the Content-Disposition header to the response 

40 # if the signed URL contains the response-content-disposition query parameter. 

41 # Cloudfront's cache policy is configured to pass this parameter through to 

42 # the origin. Otherwise, it wouldn't end up at S3. 

43 # The `ResponseContentDisposition` parameter as used in the private storage 

44 # does not work if there's no signature present. 

45 params["response-content-disposition"] = ( 

46 f'attachment; filename="{attachment}"' 

47 ) 

48 

49 return super().url(name, params, expire=expire_seconds) 

50 

51 

52class PrivateS3Storage(S3RenameMixin, S3Boto3Storage): 

53 location = settings.PRIVATE_MEDIA_LOCATION 

54 file_overwrite = False 

55 

56 def url(self, name, attachment=False, expire_seconds=None): 

57 params = {} 

58 if attachment: 

59 # Cloudfront will add the Content-Disposition header to the response 

60 # if the signed URL contains the ResponseContentDisposition query parameter. 

61 params["ResponseContentDisposition"] = ( 

62 f'attachment; filename="{attachment}"' 

63 ) 

64 

65 return super().url(name, params, expire=expire_seconds) 

66 

67 

68class StaticS3Storage(S3ManifestStaticStorage): 

69 location = settings.STATICFILES_LOCATION 

70 object_parameters = {"CacheControl": "max-age=31536000"} 

71 

72 # Clear the signing information as we don't need it for static files. 

73 # Loading the cloudfront key would waste some time for no reason. 

74 cloudfront_key_id = None 

75 cloudfront_key = None 

76 

77 

78class FileSystemRenameMixin: 

79 def rename(self, old_name, new_name): 

80 old_name = self.path(old_name) 

81 new_name = self.path(new_name) 

82 os.rename(old_name, new_name) 

83 

84 

85class PublicFileSystemStorage(FileSystemRenameMixin, FileSystemStorage): 

86 location = os.path.join(settings.MEDIA_ROOT, settings.PUBLIC_MEDIA_LOCATION) 

87 base_url = settings.PUBLIC_MEDIA_URL 

88 

89 def url(self, name, attachment=False, expire_seconds=None): 

90 return super().url(name) 

91 

92 

93class PrivateFileSystemStorage(FileSystemRenameMixin, FileSystemStorage): 

94 location = os.path.join(settings.MEDIA_ROOT, settings.PRIVATE_MEDIA_LOCATION) 

95 

96 def url( 

97 self, 

98 name, 

99 attachment=False, 

100 expire_seconds=None, 

101 ): 

102 sig_info = { 

103 "name": name, 

104 "serve_path": name, 

105 "storage": f"{self.__class__.__module__}.{self.__class__.__name__}", 

106 "attachment": attachment, 

107 } 

108 return f"{super().url(name)}?sig={signing.dumps(sig_info)}"