Skip to content

In memory

In-memory file storage implementation.

InMemoryFileStore

Bases: FileStore

In-memory file storage implementation.

This implementation stores files in memory using Python dictionaries. Files are lost when the process terminates.

Source code in dynamiq/storages/file/in_memory.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class InMemoryFileStore(FileStore):
    """In-memory file storage implementation.

    This implementation stores files in memory using Python dictionaries.
    Files are lost when the process terminates.

    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    def __init__(self, **kwargs):
        """Initialize the in-memory storage.

        Args:
            **kwargs: Additional keyword arguments (ignored)
        """
        super().__init__(**kwargs)
        self._files: dict[str, dict[str, Any]] = {}

    def list_files_bytes(self) -> list[BytesIO]:
        """List files in storage and return the content as bytes in BytesIO objects.

        Returns:
            List of BytesIO objects
        """
        files = []

        for file_path in self._files.keys():
            file = BytesIO(self._files[file_path]["content"])
            file.name = file_path
            file.description = self._files[file_path]["metadata"].get("description", "")
            file.content_type = self._files[file_path]["content_type"]
            files.append(file)

        return files

    def is_empty(self) -> bool:
        """Check if the file store is empty."""
        return len(self._files) == 0

    def store(
        self,
        file_path: str | Path,
        content: str | bytes | BinaryIO,
        content_type: str = None,
        metadata: dict[str, Any] = None,
        overwrite: bool = False,
    ) -> FileInfo:
        """Store a file in memory."""
        file_path = str(file_path)

        if file_path in self._files and not overwrite:
            logger.info(f"File '{file_path}' already exists. Skipping...")
            return self._create_file_info(file_path, self._files[file_path])

        # Convert content to bytes
        if isinstance(content, str):
            content_bytes = content.encode("utf-8")
        elif isinstance(content, bytes):
            content_bytes = content
        elif hasattr(content, "read"):  # BinaryIO-like object
            content_bytes = content.read()
            if hasattr(content, "seek"):
                content.seek(0)  # Reset position for future reads
        else:
            raise StorageError(f"Unsupported content type: {type(content)}", operation="store", path=file_path)

        if content_type is None:
            content_type, _ = mimetypes.guess_type(file_path)
            if content_type is None:
                content_type = "application/octet-stream"

        now = datetime.now()
        file_info = {
            "content": content_bytes,
            "size": len(content_bytes),
            "content_type": content_type,
            "created_at": now,
            "metadata": metadata or {},
        }

        self._files[file_path] = file_info

        return self._create_file_info(file_path, file_info)

    def retrieve(self, file_path: str | Path) -> bytes:
        """Retrieve file content as bytes."""
        file_path = str(file_path)

        if file_path not in self._files:
            raise FileNotFoundError(f"File '{file_path}' not found", operation="retrieve", path=file_path)

        return self._files[file_path]["content"]

    def exists(self, file_path: str | Path) -> bool:
        """Check if file exists."""
        return str(file_path) in self._files

    def delete(self, file_path: str | Path) -> bool:
        """Delete a file."""
        file_path = str(file_path)

        if file_path in self._files:
            del self._files[file_path]
            return True

        return False

    def list_files(
        self,
        directory: str | Path = "",
        recursive: bool = False,
    ) -> list[FileInfo]:
        """List files in storage."""
        directory = str(directory)
        files_list = []

        for file_path in self._files.keys():
            if directory and not file_path.startswith(directory):
                continue

            if not recursive:
                rel_path = file_path[len(directory) :].lstrip("/")
                if "/" in rel_path:
                    continue

            files_list.append(self._create_file_info(file_path, self._files[file_path]))

        return files_list

    def _create_file_info(self, file_path: str, file_data: dict[str, Any]) -> FileInfo:
        """Create a FileInfo object from internal file data."""
        return FileInfo(
            name=os.path.basename(file_path),
            path=file_path,
            size=file_data["size"],
            content_type=file_data["content_type"],
            created_at=file_data["created_at"],
            metadata=file_data.get("metadata", {}),
            content=file_data["content"],
        )

    def to_dict(self, **kwargs) -> dict[str, Any]:
        """Convert the InMemoryFileStore instance to a dictionary.

        Returns:
            dict: A dictionary representation of the file store that is JSON serializable.
        """

        return {
            "type": "dynamiq.storages.file.InMemoryFileStore",
            "file_count": len(self._files),
            "is_empty": self.is_empty(),
        }

__init__(**kwargs)

Initialize the in-memory storage.

Parameters:

Name Type Description Default
**kwargs

Additional keyword arguments (ignored)

 {}
Source code in dynamiq/storages/file/in_memory.py 
27
28
29
30
31
32
33
34
def __init__(self, **kwargs):
    """Initialize the in-memory storage.

    Args:
        **kwargs: Additional keyword arguments (ignored)
    """
    super().__init__(**kwargs)
    self._files: dict[str, dict[str, Any]] = {}

delete(file_path)

Delete a file.

Source code in dynamiq/storages/file/in_memory.py 
115
116
117
118
119
120
121
122
123
def delete(self, file_path: str | Path) -> bool:
    """Delete a file."""
    file_path = str(file_path)

    if file_path in self._files:
        del self._files[file_path]
        return True

    return False

exists(file_path)

Check if file exists.

Source code in dynamiq/storages/file/in_memory.py 
111
112
113
def exists(self, file_path: str | Path) -> bool:
    """Check if file exists."""
    return str(file_path) in self._files

is_empty()

Check if the file store is empty.

Source code in dynamiq/storages/file/in_memory.py 
53
54
55
def is_empty(self) -> bool:
    """Check if the file store is empty."""
    return len(self._files) == 0

list_files(directory='', recursive=False)

List files in storage.

Source code in dynamiq/storages/file/in_memory.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def list_files(
    self,
    directory: str | Path = "",
    recursive: bool = False,
) -> list[FileInfo]:
    """List files in storage."""
    directory = str(directory)
    files_list = []

    for file_path in self._files.keys():
        if directory and not file_path.startswith(directory):
            continue

        if not recursive:
            rel_path = file_path[len(directory) :].lstrip("/")
            if "/" in rel_path:
                continue

        files_list.append(self._create_file_info(file_path, self._files[file_path]))

    return files_list

list_files_bytes()

List files in storage and return the content as bytes in BytesIO objects.

Returns:

Type Description
list[BytesIO]

List of BytesIO objects
Source code in dynamiq/storages/file/in_memory.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def list_files_bytes(self) -> list[BytesIO]:
    """List files in storage and return the content as bytes in BytesIO objects.

    Returns:
        List of BytesIO objects
    """
    files = []

    for file_path in self._files.keys():
        file = BytesIO(self._files[file_path]["content"])
        file.name = file_path
        file.description = self._files[file_path]["metadata"].get("description", "")
        file.content_type = self._files[file_path]["content_type"]
        files.append(file)

    return files

retrieve(file_path)

Retrieve file content as bytes.

Source code in dynamiq/storages/file/in_memory.py 
102
103
104
105
106
107
108
109
def retrieve(self, file_path: str | Path) -> bytes:
    """Retrieve file content as bytes."""
    file_path = str(file_path)

    if file_path not in self._files:
        raise FileNotFoundError(f"File '{file_path}' not found", operation="retrieve", path=file_path)

    return self._files[file_path]["content"]

store(file_path, content, content_type=None, metadata=None, overwrite=False)

Store a file in memory.

Source code in dynamiq/storages/file/in_memory.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def store(
    self,
    file_path: str | Path,
    content: str | bytes | BinaryIO,
    content_type: str = None,
    metadata: dict[str, Any] = None,
    overwrite: bool = False,
) -> FileInfo:
    """Store a file in memory."""
    file_path = str(file_path)

    if file_path in self._files and not overwrite:
        logger.info(f"File '{file_path}' already exists. Skipping...")
        return self._create_file_info(file_path, self._files[file_path])

    # Convert content to bytes
    if isinstance(content, str):
        content_bytes = content.encode("utf-8")
    elif isinstance(content, bytes):
        content_bytes = content
    elif hasattr(content, "read"):  # BinaryIO-like object
        content_bytes = content.read()
        if hasattr(content, "seek"):
            content.seek(0)  # Reset position for future reads
    else:
        raise StorageError(f"Unsupported content type: {type(content)}", operation="store", path=file_path)

    if content_type is None:
        content_type, _ = mimetypes.guess_type(file_path)
        if content_type is None:
            content_type = "application/octet-stream"

    now = datetime.now()
    file_info = {
        "content": content_bytes,
        "size": len(content_bytes),
        "content_type": content_type,
        "created_at": now,
        "metadata": metadata or {},
    }

    self._files[file_path] = file_info

    return self._create_file_info(file_path, file_info)

to_dict(**kwargs)

Convert the InMemoryFileStore instance to a dictionary.

Returns:

Name Type Description
dict dict[str, Any]

A dictionary representation of the file store that is JSON serializable.
Source code in dynamiq/storages/file/in_memory.py 
159
160
161
162
163
164
165
166
167
168
169
170
def to_dict(self, **kwargs) -> dict[str, Any]:
    """Convert the InMemoryFileStore instance to a dictionary.

    Returns:
        dict: A dictionary representation of the file store that is JSON serializable.
    """

    return {
        "type": "dynamiq.storages.file.InMemoryFileStore",
        "file_count": len(self._files),
        "is_empty": self.is_empty(),
    }