Coverage for src/backend/InvenTree/InvenTree/setting/storages.py: 63%

1"""Settings for storage backends.""" 

2 

3from enum import Enum 

4from typing import Optional 

5 

6from InvenTree.config import get_boolean_setting, get_setting 

7 

8 

9class StorageBackends(str, Enum): 

10 """Enumeration of available storage backends.""" 

11 

12 LOCAL = 'local' 

13 S3 = 's3' 

14 SFTP = 'sftp' 

15 

16 

17STORAGE_BACKEND_MAPPING = { 

18 StorageBackends.LOCAL: 'django.core.files.storage.FileSystemStorage', 

19 StorageBackends.S3: 'storages.backends.s3.S3Storage', 

20 StorageBackends.SFTP: 'storages.backends.sftpstorage.SFTPStorage', 

21} 

22 

23 

24def init_storages() -> tuple[str, dict, Optional[str]]: 

25 """Initialize storage backend settings.""" 

26 target = get_setting( 

27 'INVENTREE_STORAGE_TARGET', 

28 'storage.target', 

29 StorageBackends.LOCAL.value, 

30 typecast=str, 

31 ) 

32 

33 # Check that the target is valid 

34 if target not in STORAGE_BACKEND_MAPPING: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 raise ValueError(f"Invalid storage target: '{target}'") 

36 

37 options = {} 

38 media_url: Optional[str] = None 

39 if target == StorageBackends.S3.value: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 s3_bucket = get_setting( 

41 'INVENTREE_S3_BUCKET_NAME', 'storage.s3.bucket_name', None, typecast=str 

42 ) 

43 s3_acl = get_setting( 

44 'INVENTREE_S3_DEFAULT_ACL', 'storage.s3.default_acl', None, typecast=str 

45 ) 

46 s3_endpoint = get_setting( 

47 'INVENTREE_S3_ENDPOINT_URL', 'storage.s3.endpoint_url', None, typecast=str 

48 ) 

49 s3_location = get_setting( 

50 'INVENTREE_S3_LOCATION', 

51 'storage.s3.location', 

52 'inventree-server', 

53 typecast=str, 

54 ) 

55 

56 media_url = f'{s3_endpoint}/{s3_bucket}/{s3_location}/' 

57 options = { 

58 'access_key': get_setting( 

59 'INVENTREE_S3_ACCESS_KEY', 'storage.s3.access_key', None, typecast=str 

60 ), 

61 'secret_key': get_setting( 

62 'INVENTREE_S3_SECRET_KEY', 'storage.s3.secret_key', None, typecast=str 

63 ), 

64 'bucket_name': s3_bucket, 

65 'default_acl': s3_acl, 

66 'region_name': get_setting( 

67 'INVENTREE_S3_REGION_NAME', 'storage.s3.region_name', None, typecast=str 

68 ), 

69 'endpoint_url': s3_endpoint, 

70 'verify': get_boolean_setting( 

71 'INVENTREE_S3_VERIFY_SSL', 'storage.s3.verify_ssl', True 

72 ), 

73 'location': s3_location, 

74 'file_overwrite': get_boolean_setting( 

75 'INVENTREE_S3_OVERWRITE', 'storage.s3.overwrite', True 

76 ), 

77 'addressing_style': 'virtual' 

78 if get_boolean_setting('INVENTREE_S3_VIRTUAL', 'storage.s3.virtual', False) 

79 else 'path', 

80 'object_parameters': {'CacheControl': 'public,max-age=86400'}, 

81 } 

82 elif target == StorageBackends.SFTP: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 options = { 

84 'host': get_setting('INVENTREE_SFTP_HOST', 'sftp.host', None, typecast=str), 

85 'uid': get_setting('INVENTREE_SFTP_UID', 'sftp.uid', None, typecast=int), 

86 'gid': get_setting('INVENTREE_SFTP_GID', 'sftp.gid', None, typecast=int), 

87 'location': get_setting( 

88 'INVENTREE_SFTP_LOCATION', 

89 'sftp.location', 

90 'inventree-server', 

91 typecast=str, 

92 ), 

93 'params': get_setting( 

94 'INVENTREE_SFTP_PARAMS', 'sftp.params', {}, typecast=dict 

95 ), 

96 } 

97 return ( 

98 target, 

99 { 

100 'default': { 

101 'BACKEND': STORAGE_BACKEND_MAPPING.get( 

102 target, STORAGE_BACKEND_MAPPING[StorageBackends.LOCAL] 

103 ), 

104 'OPTIONS': options, 

105 }, 

106 'staticfiles': { 

107 'BACKEND': 'whitenoise.storage.CompressedStaticFilesStorage' 

108 }, 

109 }, 

110 media_url, 

111 )