Coverage for baibu/core/storage.py: 100%

1"""One interface for contributor files, whatever the backend. 

2 

3Submissions, audio and translations store files through these functions 

4rather than calling a storage SDK. The backend (local filesystem or any 

5S3-compatible store) is chosen by ``STORAGE_BACKEND``; see 

6:mod:`baibu.core.storage_config`. 

7 

8Records keep the returned *key* (a relative path such as 

9``submissions/raw/2026/10/06/<id>.json``), never a backend-specific URI, so 

10a deployment can move its files to another backend without rewriting rows. 

11""" 

12 

13import json 

14import logging 

15from datetime import UTC 

16from datetime import datetime 

17 

18from django.core.files.base import ContentFile 

19from django.core.files.storage import FileSystemStorage 

20from django.core.files.storage import Storage 

21from django.core.files.storage import storages 

22 

23logger = logging.getLogger(__name__) 

24 

25PRIVATE = "private" 

26 

27 

28class PrivateFileSystemStorage(FileSystemStorage): 

29 """Filesystem storage whose files have no URL.""" 

30 

31 def url(self, name): 

32 msg = "Private files are not served over HTTP." 

33 raise NotImplementedError(msg) 

34 

35 

36def private_storage() -> Storage: 

37 return storages[PRIVATE] 

38 

39 

40def build_key(area: str, name: str, *, now: datetime | None = None) -> str: 

41 """Return ``<area>/<yyyy>/<mm>/<dd>/<name>``. 

42 

43 ``area`` is a path such as ``submissions/raw``; ``name`` should be unique, 

44 for example a record's UUID with an extension. 

45 """ 

46 now = now or datetime.now(UTC) 

47 area = area.strip("/") 

48 if not area or not name or "/" in name or name in {".", ".."}: 

49 msg = f"Invalid storage key parts: area={area!r} name={name!r}" 

50 raise ValueError(msg) 

51 return f"{area}/{now:%Y/%m/%d}/{name}" 

52 

53 

54def save_bytes(key: str, data: bytes) -> str: 

55 """Store ``data`` under ``key`` and return the key actually used. 

56 

57 If the key is already taken the backend picks a free variant, so always 

58 keep the returned value. 

59 """ 

60 saved = private_storage().save(key, ContentFile(data)) 

61 logger.info("Stored %d bytes at %s", len(data), saved) 

62 return saved 

63 

64 

65def save_json(key: str, payload: dict) -> str: 

66 return save_bytes(key, json.dumps(payload, ensure_ascii=False, indent=2).encode()) 

67 

68 

69def read_bytes(key: str) -> bytes | None: 

70 """Return the file's contents, or ``None`` if there is no such file.""" 

71 if not key: 

72 return None 

73 storage = private_storage() 

74 try: 

75 with storage.open(key, "rb") as handle: 

76 return handle.read() 

77 except FileNotFoundError: 

78 return None 

79 

80 

81def read_json(key: str) -> dict | None: 

82 data = read_bytes(key) 

83 return None if data is None else json.loads(data) 

84 

85 

86def exists(key: str) -> bool: 

87 return bool(key) and private_storage().exists(key) 

88 

89 

90def delete(key: str) -> bool: 

91 """Delete the file. Returns ``False`` if it was already gone.""" 

92 if not exists(key): 

93 return False 

94 private_storage().delete(key) 

95 logger.info("Deleted %s", key) 

96 return True