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

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

__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 
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

is_empty()

Check if the file store is empty.

Source code in dynamiq/storages/file/in_memory.py 
58
59
60
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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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 
 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
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)