Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/import_export/tmp_storages.py: 51%
60 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1import os
2import tempfile
3from uuid import uuid4
5from django.core.cache import cache
6from django.core.files.base import ContentFile
7from django.core.files.storage import default_storage
10class BaseStorage:
12 def __init__(self, name=None, read_mode='r', encoding=None):
13 self.name = name
14 self.read_mode = read_mode
15 self.encoding = encoding
17 def save(self, data):
18 raise NotImplementedError
20 def read(self):
21 raise NotImplementedError
23 def remove(self):
24 raise NotImplementedError
27class TempFolderStorage(BaseStorage):
29 def save(self, data):
30 with self._open(mode='w') as file:
31 file.write(data)
33 def read(self):
34 with self._open(mode=self.read_mode) as file:
35 return file.read()
37 def remove(self):
38 os.remove(self.get_full_path())
40 def get_full_path(self):
41 return os.path.join(
42 tempfile.gettempdir(),
43 self.name
44 )
46 def _open(self, mode='r'):
47 if self.name:
48 return open(self.get_full_path(), mode, encoding=self.encoding)
49 else:
50 tmp_file = tempfile.NamedTemporaryFile(delete=False)
51 self.name = tmp_file.name
52 return tmp_file
55class CacheStorage(BaseStorage):
56 """
57 By default memcache maximum size per key is 1MB, be careful with large files.
58 """
59 CACHE_LIFETIME = 86400
60 CACHE_PREFIX = 'django-import-export-'
62 def save(self, data):
63 if not self.name:
64 self.name = uuid4().hex
65 cache.set(self.CACHE_PREFIX + self.name, data, self.CACHE_LIFETIME)
67 def read(self):
68 return cache.get(self.CACHE_PREFIX + self.name)
70 def remove(self):
71 cache.delete(self.name)
74class MediaStorage(BaseStorage):
75 MEDIA_FOLDER = 'django-import-export'
77 def __init__(self, name=None, read_mode='rb', encoding=None):
78 super().__init__(name, read_mode=read_mode, encoding=encoding)
80 def save(self, data):
81 if not self.name:
82 self.name = uuid4().hex
83 default_storage.save(self.get_full_path(), ContentFile(data))
85 def read(self):
86 with default_storage.open(self.get_full_path(), mode=self.read_mode) as f:
87 return f.read()
89 def remove(self):
90 default_storage.delete(self.get_full_path())
92 def get_full_path(self):
93 return os.path.join(
94 self.MEDIA_FOLDER,
95 self.name
96 )