from os import makedirs, remove
from os.path import abspath, join, dirname, exists
from .base import Store
from sqlalchemy_media.constants import KB
from sqlalchemy_media.helpers import copy_stream
from sqlalchemy_media.typing_ import FileLike
[docs]class FileSystemStore(Store):
"""
Store for dealing with local file-system.
:param root_path: The path to a directory to store files.
:param base_url: The base url path to include at the beginning of the
file's path to yield the access url.
:param chunk_size: Length of the chunks to read/write from/to files.
default: 32768 = 32 * 1024
"""
def __init__(self, root_path: str, base_url: str,
chunk_size: int = 32 * KB):
self.root_path = abspath(root_path)
self.base_url = base_url.rstrip('/')
self.chunk_size = chunk_size
def _get_physical_path(self, filename: str) -> str:
return join(self.root_path, filename)
[docs] def put(self, filename: str, stream: FileLike):
physical_path = self._get_physical_path(filename)
physical_directory = dirname(physical_path)
if not exists(physical_directory):
makedirs(physical_directory, exist_ok=True)
with open(physical_path, mode='wb') as target_file:
return copy_stream(
stream,
target_file,
chunk_size=self.chunk_size
)
[docs] def delete(self, filename: str):
physical_path = self._get_physical_path(filename)
remove(physical_path)
[docs] def open(self, filename: str, mode: str = 'rb') -> FileLike:
return open(self._get_physical_path(filename), mode=mode)
[docs] def locate(self, attachment) -> str:
return '%s/%s' % (self.base_url, attachment.path)