managers.file_manager 

 1import json
 2from pathlib import Path
 3from typing import Any
 4
 5import aiofiles
 6import aiofiles.os
 7
 8
 9class AsyncFileManager:
10    @staticmethod
11    async def get_json_files(input_path: Path) -> list[Path]:
12        """
13        Recursively retrieves all JSON files located in nested subdirectories of the input path.
14
15        Note:
16            The input directory may contain multiple folders with arbitrary names
17            (e.g. per day), and JSON files are expected to be inside those folders.
18
19        :param input_path: The INPUT directory path where folders are placed.
20        :return: List of Path objects pointing to JSON files.
21        """
22        return list(input_path.rglob("*.json"))
23
24    @staticmethod
25    async def read_json(path: Path) -> dict[str, Any]:
26        """
27        Asynchronously reads and parses a JSON file.
28
29        :param path: The path to the JSON file.
30        :return: The file content as a dictionary.
31        """
32        async with aiofiles.open(path, "r", encoding="utf-8") as f:
33            content = await f.read()
34            return json.loads(content)
35
36    @staticmethod
37    async def write_json(path: Path, data: dict[str, Any]) -> None:
38        """
39        Asynchronously writes a dictionary to a file in JSON format.
40
41        :param path: The file path to write to.
42        :param data: The dictionary to serialize and write.
43        """
44        async with aiofiles.open(path, "w", encoding="utf-8") as f:
45            await f.write(json.dumps(data, indent=2, ensure_ascii=False))
46
47    @staticmethod
48    async def delete_file(path: Path) -> None:
49        """
50        Asynchronously deletes a file from the filesystem.
51
52        :param path: The path of the file to delete.
53        """
54        await aiofiles.os.remove(path)
class AsyncFileManager: 
10class AsyncFileManager:
11    @staticmethod
12    async def get_json_files(input_path: Path) -> list[Path]:
13        """
14        Recursively retrieves all JSON files located in nested subdirectories of the input path.
15
16        Note:
17            The input directory may contain multiple folders with arbitrary names
18            (e.g. per day), and JSON files are expected to be inside those folders.
19
20        :param input_path: The INPUT directory path where folders are placed.
21        :return: List of Path objects pointing to JSON files.
22        """
23        return list(input_path.rglob("*.json"))
24
25    @staticmethod
26    async def read_json(path: Path) -> dict[str, Any]:
27        """
28        Asynchronously reads and parses a JSON file.
29
30        :param path: The path to the JSON file.
31        :return: The file content as a dictionary.
32        """
33        async with aiofiles.open(path, "r", encoding="utf-8") as f:
34            content = await f.read()
35            return json.loads(content)
36
37    @staticmethod
38    async def write_json(path: Path, data: dict[str, Any]) -> None:
39        """
40        Asynchronously writes a dictionary to a file in JSON format.
41
42        :param path: The file path to write to.
43        :param data: The dictionary to serialize and write.
44        """
45        async with aiofiles.open(path, "w", encoding="utf-8") as f:
46            await f.write(json.dumps(data, indent=2, ensure_ascii=False))
47
48    @staticmethod
49    async def delete_file(path: Path) -> None:
50        """
51        Asynchronously deletes a file from the filesystem.
52
53        :param path: The path of the file to delete.
54        """
55        await aiofiles.os.remove(path)
@staticmethod
async def get_json_files(input_path: pathlib.Path) -> list[pathlib.Path]: 
11    @staticmethod
12    async def get_json_files(input_path: Path) -> list[Path]:
13        """
14        Recursively retrieves all JSON files located in nested subdirectories of the input path.
15
16        Note:
17            The input directory may contain multiple folders with arbitrary names
18            (e.g. per day), and JSON files are expected to be inside those folders.
19
20        :param input_path: The INPUT directory path where folders are placed.
21        :return: List of Path objects pointing to JSON files.
22        """
23        return list(input_path.rglob("*.json"))

Recursively retrieves all JSON files located in nested subdirectories of the input path.

Note: The input directory may contain multiple folders with arbitrary names (e.g. per day), and JSON files are expected to be inside those folders.

Parameters
  • input_path: The INPUT directory path where folders are placed.
Returns

List of Path objects pointing to JSON files.

@staticmethod
async def read_json(path: pathlib.Path) -> dict[str, typing.Any]: 
25    @staticmethod
26    async def read_json(path: Path) -> dict[str, Any]:
27        """
28        Asynchronously reads and parses a JSON file.
29
30        :param path: The path to the JSON file.
31        :return: The file content as a dictionary.
32        """
33        async with aiofiles.open(path, "r", encoding="utf-8") as f:
34            content = await f.read()
35            return json.loads(content)

Asynchronously reads and parses a JSON file.

Parameters
  • path: The path to the JSON file.
Returns

The file content as a dictionary.

@staticmethod
async def write_json(path: pathlib.Path, data: dict[str, typing.Any]) -> None: 
37    @staticmethod
38    async def write_json(path: Path, data: dict[str, Any]) -> None:
39        """
40        Asynchronously writes a dictionary to a file in JSON format.
41
42        :param path: The file path to write to.
43        :param data: The dictionary to serialize and write.
44        """
45        async with aiofiles.open(path, "w", encoding="utf-8") as f:
46            await f.write(json.dumps(data, indent=2, ensure_ascii=False))

Asynchronously writes a dictionary to a file in JSON format.

Parameters
  • path: The file path to write to.
  • data: The dictionary to serialize and write.
@staticmethod
async def delete_file(path: pathlib.Path) -> None: 
48    @staticmethod
49    async def delete_file(path: Path) -> None:
50        """
51        Asynchronously deletes a file from the filesystem.
52
53        :param path: The path of the file to delete.
54        """
55        await aiofiles.os.remove(path)

Asynchronously deletes a file from the filesystem.

Parameters
  • path: The path of the file to delete.