Coverage for file_utils.py: 100%

1import aiofiles 

2import aiofiles.os 

3import base64 

4 

5 

6def binary_to_base64(binary_data: bytes) -> str: 

7 """Converts binary data to a base64-encoded string.""" 

8 if not isinstance(binary_data, bytes): 

9 raise TypeError(f"Expected bytes for binary_data, got {type(binary_data)}") 

10 base64_bytes = base64.b64encode(binary_data) 

11 base64_str = base64_bytes.decode('utf-8') 

12 return base64_str 

13 

14 

15def base64_to_binary(base64_str: str) -> bytes: 

16 """Converts a base64-encoded string to binary data.""" 

17 if not isinstance(base64_str, str): 

18 raise TypeError(f"Expected str for base64_str, got {type(base64_str)}") 

19 base64_bytes = base64_str.encode('utf-8') 

20 binary_data = base64.b64decode(base64_bytes) 

21 return binary_data 

22 

23 

24async def save_base64_as_binary( 

25 file_path: str, 

26 base64_str: str 

27) -> str: 

28 assert isinstance(file_path, str) 

29 assert isinstance(base64_str, str) 

30 binary_data = base64_to_binary(base64_str) 

31 async with aiofiles.open(file_path, "wb") as file: 

32 await file.write(binary_data) 

33 return file_path 

34 

35 

36async def read_file_bytes( 

37 file_path: str 

38) -> bytes: 

39 """Read a file asynchronously and return its content as bytes.""" 

40 if not isinstance(file_path, str): 

41 raise TypeError(f"Expected str for file_path, got {type(file_path)}") 

42 if not await aiofiles.os.path.exists(file_path): 

43 raise FileNotFoundError(f"File does not exist: {file_path}") 

44 async with aiofiles.open(file_path, "rb") as file: 

45 content = await file.read() 

46 return content 

47 

48 

49async def read_file_base64( 

50 file_path: str 

51) -> str: 

52 """Read a file asynchronously and return its content as a base64-encoded string.""" 

53 file_bytes = await read_file_bytes(file_path) 

54 return binary_to_base64(file_bytes)