Source code for

Based on dj-inmemorystorage (BSD) by Cody Soyland, Seán Hayes, Tore Birkeland,
and Nick Presta.

import errno
import io
import os
import pathlib
from urllib.parse import urljoin

from django.conf import settings
from django.core.files.base import ContentFile
from django.core.signals import setting_changed
from django.utils._os import safe_join
from django.utils.deconstruct import deconstructible
from django.utils.encoding import filepath_to_uri
from django.utils.functional import cached_property
from django.utils.timezone import now

from .base import Storage
from .mixins import StorageSettingsMixin

__all__ = ("InMemoryStorage",)

class TimingMixin:
    def _initialize_times(self):
        self.created_time = now()
        self.accessed_time = self.created_time
        self.modified_time = self.created_time

    def _update_accessed_time(self):
        self.accessed_time = now()

    def _update_modified_time(self):
        self.modified_time = now()

class InMemoryFileNode(ContentFile, TimingMixin):
    Helper class representing an in-memory file node.

    Handle unicode/bytes conversion during I/O operations and record creation,
    modification, and access times.

    def __init__(self, content="", name=""):
        self.file = None
        self._content_type = type(content)

    def open(self, mode):
        return super().open(mode)

    def write(self, data):

    def _initialize_stream(self):
        """Initialize underlying stream according to the content type."""
        self.file = io.BytesIO() if self._content_type == bytes else io.StringIO()

    def _convert_stream_content(self, mode):
        """Convert actual file content according to the opening mode."""
        new_content_type = bytes if "b" in mode else str
        # No conversion needed.
        if self._content_type == new_content_type:

        content = self.file.getvalue()
        content = content.encode() if isinstance(content, str) else content.decode()
        self._content_type = new_content_type


class InMemoryDirNode(TimingMixin):
    Helper class representing an in-memory directory node.

    Handle path navigation of directory trees, creating missing nodes if

    def __init__(self):
        self._children = {}

    def resolve(self, path, create_if_missing=False, leaf_cls=None, check_exists=True):
        Navigate current directory tree, returning node matching path or
        creating a new one, if missing.
        - path: path of the node to search
        - create_if_missing: create nodes if not exist. Defaults to False.
        - leaf_cls: expected type of leaf node. Defaults to None.
        - check_exists: if True and the leaf node does not exist, raise a
          FileNotFoundError. Defaults to True.
        path_segments = list(pathlib.Path(path).parts)
        current_node = self

        while path_segments:
            path_segment = path_segments.pop(0)
            # If current node is a file node and there are unprocessed
            # segments, raise an error.
            if isinstance(current_node, InMemoryFileNode):
                path_segments = os.path.split(path)
                current_path = "/".join(
                    path_segments[: path_segments.index(path_segment)]
                raise NotADirectoryError(
                    errno.ENOTDIR, os.strerror(errno.ENOTDIR), current_path
            current_node = current_node._resolve_child(
                leaf_cls if len(path_segments) == 0 else InMemoryDirNode,
            if current_node is None:

        if current_node is None and check_exists:
            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), path)

        # If a leaf_cls is not None, check if leaf node is of right type.
        if leaf_cls and not isinstance(current_node, leaf_cls):
            error_cls, error_code = (
                (NotADirectoryError, errno.ENOTDIR)
                if leaf_cls is InMemoryDirNode
                else (IsADirectoryError, errno.EISDIR)
            raise error_cls(error_code, os.strerror(error_code), path)

        return current_node

    def _resolve_child(self, path_segment, create_if_missing, child_cls):
        if create_if_missing:
            return self._children.setdefault(path_segment, child_cls())
        return self._children.get(path_segment)

    def listdir(self):
        directories, files = [], []
        for name, entry in self._children.items():
            if isinstance(entry, InMemoryDirNode):
        return directories, files

    def remove_child(self, name):
        if name in self._children:
            del self._children[name]

[docs]@deconstructible(path="") class InMemoryStorage(Storage, StorageSettingsMixin): """A storage saving files in memory.""" def __init__( self, location=None, base_url=None, file_permissions_mode=None, directory_permissions_mode=None, ): self._location = location self._base_url = base_url self._file_permissions_mode = file_permissions_mode self._directory_permissions_mode = directory_permissions_mode self._root = InMemoryDirNode() self._resolve( self.base_location, create_if_missing=True, leaf_cls=InMemoryDirNode ) setting_changed.connect(self._clear_cached_properties) @cached_property def base_location(self): return self._value_or_setting(self._location, settings.MEDIA_ROOT) @cached_property def location(self): return os.path.abspath(self.base_location) @cached_property def base_url(self): if self._base_url is not None and not self._base_url.endswith("/"): self._base_url += "/" return self._value_or_setting(self._base_url, settings.MEDIA_URL) @cached_property def file_permissions_mode(self): return self._value_or_setting( self._file_permissions_mode, settings.FILE_UPLOAD_PERMISSIONS ) @cached_property def directory_permissions_mode(self): return self._value_or_setting( self._directory_permissions_mode, settings.FILE_UPLOAD_DIRECTORY_PERMISSIONS ) def _relative_path(self, name): full_path = self.path(name) return os.path.relpath(full_path, self.location) def _resolve(self, name, create_if_missing=False, leaf_cls=None, check_exists=True): try: relative_path = self._relative_path(name) return self._root.resolve( relative_path, create_if_missing=create_if_missing, leaf_cls=leaf_cls, check_exists=check_exists, ) except NotADirectoryError as exc: absolute_path = self.path(exc.filename) raise FileExistsError(f"{absolute_path} exists and is not a directory.") def _open(self, name, mode="rb"): create_if_missing = "w" in mode file_node = self._resolve( name, create_if_missing=create_if_missing, leaf_cls=InMemoryFileNode ) return def _save(self, name, content): file_node = self._resolve( name, create_if_missing=True, leaf_cls=InMemoryFileNode ) fd = None for chunk in content.chunks(): if fd is None: mode = "wb" if isinstance(chunk, bytes) else "wt" fd = fd.write(chunk) if hasattr(content, "temporary_file_path"): os.remove(content.temporary_file_path()) file_node.modified_time = now() return self._relative_path(name).replace("\\", "/") def path(self, name): return safe_join(self.location, name) def delete(self, name): path, filename = os.path.split(name) dir_node = self._resolve(path, check_exists=False) if dir_node is None: return None dir_node.remove_child(filename) def exists(self, name): return self._resolve(name, check_exists=False) is not None def listdir(self, path): node = self._resolve(path, leaf_cls=InMemoryDirNode) return node.listdir() def size(self, name): return len(self._open(name, "rb").file.getvalue()) def url(self, name): if self.base_url is None: raise ValueError("This file is not accessible via a URL.") url = filepath_to_uri(name) if url is not None: url = url.lstrip("/") return urljoin(self.base_url, url) def get_accessed_time(self, name): file_node = self._resolve(name) return file_node.accessed_time def get_created_time(self, name): file_node = self._resolve(name) return file_node.created_time def get_modified_time(self, name): file_node = self._resolve(name) return file_node.modified_time
Back to Top