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 
 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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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, file_paths: list[str] | None = None) -> list[BytesIO]:
        """Return stored files as BytesIO objects.

        Args:
            file_paths: If provided, return only these files. Otherwise return all files.

        Returns:
            List of BytesIO objects with name, description, and content_type attributes.
        """
        keys = file_paths if file_paths else list(self._files.keys())
        files = []
        for file_path in keys:
            if file_path not in self._files:
                continue
            data = self._files[file_path]
            file = BytesIO(data["content"])
            file.name = file_path
            file.description = data["metadata"].get("description", "")
            file.content_type = data["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])

        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 to_checkpoint_state(self, max_bytes: int, file_paths: Collection[str] | None = None) -> dict[str, Any]:
        """Return the stored files in a JSON-safe form, for a checkpoint to carry.

        Only the files at ``file_paths`` are taken when it is given. Files are taken in the order
        they were stored while their contents fit in ``max_bytes``. A file that does not fit is
        left out with a warning, so the checkpoint stays small enough to save.
        """
        files: dict[str, dict[str, Any]] = {}
        skipped: list[str] = []
        total = 0
        for file_path, file_data in self._files.items():
            if file_paths is not None and file_path not in file_paths:
                continue
            if total + file_data["size"] > max_bytes:
                skipped.append(file_path)
                continue
            total += file_data["size"]
            files[file_path] = {
                "content": base64.b64encode(file_data["content"]).decode("ascii"),
                "content_type": file_data["content_type"],
                "created_at": file_data["created_at"].isoformat(),
                "metadata": file_data["metadata"],
            }
        if skipped:
            logger.warning(
                f"InMemoryFileStore: {len(skipped)} file(s) do not fit the {max_bytes}-byte checkpoint "
                f"budget and are not saved: {skipped}"
            )
        return {"files": files}

    def from_checkpoint_state(self, state: dict[str, Any]) -> None:
        """Store the files saved by ``to_checkpoint_state``, replacing any file at the same path."""
        for file_path, file_data in (state.get("files") or {}).items():
            content = base64.b64decode(file_data["content"])
            self._files[file_path] = {
                "content": content,
                "size": len(content),
                "content_type": file_data["content_type"],
                "created_at": datetime.fromisoformat(file_data["created_at"]),
                "metadata": file_data["metadata"],
            }

    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"],
        )

__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 
28
29
30
31
32
33
34
35
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 
120
121
122
123
124
125
126
127
128
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 
116
117
118
def exists(self, file_path: str | Path) -> bool:
    """Check if file exists."""
    return str(file_path) in self._files

from_checkpoint_state(state)

Store the files saved by to_checkpoint_state, replacing any file at the same path.

Source code in dynamiq/storages/file/in_memory.py 
182
183
184
185
186
187
188
189
190
191
192
def from_checkpoint_state(self, state: dict[str, Any]) -> None:
    """Store the files saved by ``to_checkpoint_state``, replacing any file at the same path."""
    for file_path, file_data in (state.get("files") or {}).items():
        content = base64.b64decode(file_data["content"])
        self._files[file_path] = {
            "content": content,
            "size": len(content),
            "content_type": file_data["content_type"],
            "created_at": datetime.fromisoformat(file_data["created_at"]),
            "metadata": file_data["metadata"],
        }

is_empty()

Check if the file store is empty.

Source code in dynamiq/storages/file/in_memory.py 
59
60
61
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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(file_paths=None)

Return stored files as BytesIO objects.

Parameters:

Name Type Description Default
file_paths list[str] | None

If provided, return only these files. Otherwise return all files.

 None

Returns:

Type Description
list[BytesIO]

List of BytesIO objects with name, description, and content_type attributes.
Source code in dynamiq/storages/file/in_memory.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def list_files_bytes(self, file_paths: list[str] | None = None) -> list[BytesIO]:
    """Return stored files as BytesIO objects.

    Args:
        file_paths: If provided, return only these files. Otherwise return all files.

    Returns:
        List of BytesIO objects with name, description, and content_type attributes.
    """
    keys = file_paths if file_paths else list(self._files.keys())
    files = []
    for file_path in keys:
        if file_path not in self._files:
            continue
        data = self._files[file_path]
        file = BytesIO(data["content"])
        file.name = file_path
        file.description = data["metadata"].get("description", "")
        file.content_type = data["content_type"]
        files.append(file)
    return files

retrieve(file_path)

Retrieve file content as bytes.

Source code in dynamiq/storages/file/in_memory.py 
107
108
109
110
111
112
113
114
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 
 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
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])

    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_checkpoint_state(max_bytes, file_paths=None)

Return the stored files in a JSON-safe form, for a checkpoint to carry.

Only the files at file_paths are taken when it is given. Files are taken in the order they were stored while their contents fit in max_bytes. A file that does not fit is left out with a warning, so the checkpoint stays small enough to save.

Source code in dynamiq/storages/file/in_memory.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def to_checkpoint_state(self, max_bytes: int, file_paths: Collection[str] | None = None) -> dict[str, Any]:
    """Return the stored files in a JSON-safe form, for a checkpoint to carry.

    Only the files at ``file_paths`` are taken when it is given. Files are taken in the order
    they were stored while their contents fit in ``max_bytes``. A file that does not fit is
    left out with a warning, so the checkpoint stays small enough to save.
    """
    files: dict[str, dict[str, Any]] = {}
    skipped: list[str] = []
    total = 0
    for file_path, file_data in self._files.items():
        if file_paths is not None and file_path not in file_paths:
            continue
        if total + file_data["size"] > max_bytes:
            skipped.append(file_path)
            continue
        total += file_data["size"]
        files[file_path] = {
            "content": base64.b64encode(file_data["content"]).decode("ascii"),
            "content_type": file_data["content_type"],
            "created_at": file_data["created_at"].isoformat(),
            "metadata": file_data["metadata"],
        }
    if skipped:
        logger.warning(
            f"InMemoryFileStore: {len(skipped)} file(s) do not fit the {max_bytes}-byte checkpoint "
            f"budget and are not saved: {skipped}"
        )
    return {"files": files}