Coverage for whole_app/dictionaries/file.py: 100%

1import typing 

2 

3from anyio import Path as AsyncPath 

4 

5from whole_app.settings import SETTINGS 

6 

7 

8def init_storage() -> None: 

9 SETTINGS.dictionaries_path.mkdir( 

10 parents=True, 

11 exist_ok=True, 

12 ) 

13 

14 

15class FileProvider: 

16 _user_dict_path: AsyncPath 

17 

18 def prepare(self: "FileProvider", user_name: str) -> "FileProvider": 

19 self._user_dict_path = AsyncPath(SETTINGS.dictionaries_path / user_name) 

20 return self 

21 

22 async def _store_lines(self: "FileProvider", lines: list[str]) -> None: 

23 await self._user_dict_path.write_text("\n".join(lines) + "\n") 

24 

25 async def save_record(self: "FileProvider", exception_word: str) -> None: 

26 await self._user_dict_path.touch() 

27 clean_word: typing.Final = exception_word.strip().lower() 

28 file_content: typing.Final = await self.fetch_records() 

29 if clean_word not in file_content: 

30 file_content.append(clean_word) 

31 await self._store_lines(file_content) 

32 

33 async def remove_record(self: "FileProvider", exception_word: str) -> None: 

34 file_content: typing.Final = await self.fetch_records() 

35 if exception_word in file_content: 

36 file_content.remove(exception_word) 

37 await self._store_lines(file_content) 

38 

39 async def fetch_records(self: "FileProvider") -> list[str]: 

40 if await self._user_dict_path.exists(): 

41 return [one_line.strip() for one_line in (await self._user_dict_path.read_text()).split("\n") if one_line] 

42 return []