(self, item: Dict, item_type: str)
| 54 | await writer.writerow(item) |
| 55 | |
| 56 | async def write_to_jsonl(self, item: Dict, item_type: str): |
| 57 | file_path = self._get_file_path('jsonl', item_type) |
| 58 | async with self.lock: |
| 59 | async with aiofiles.open(file_path, 'a', encoding='utf-8') as f: |
| 60 | await f.write(json.dumps(item, ensure_ascii=False) + '\n') |
| 61 | |
| 62 | async def write_single_item_to_json(self, item: Dict, item_type: str): |
| 63 | file_path = self._get_file_path('json', item_type) |
no test coverage detected