Writes an iterable of dictionaries to jsonl
(
filename: str, data: Iterable[Dict], append: bool = False, drop_builtin: bool = True
)
| 46 | |
| 47 | |
| 48 | def write_jsonl( |
| 49 | filename: str, data: Iterable[Dict], append: bool = False, drop_builtin: bool = True |
| 50 | ): |
| 51 | """ |
| 52 | Writes an iterable of dictionaries to jsonl |
| 53 | """ |
| 54 | if append: |
| 55 | mode = "ab" |
| 56 | else: |
| 57 | mode = "wb" |
| 58 | filename = os.path.expanduser(filename) |
| 59 | if filename.endswith(".gz"): |
| 60 | with open(filename, mode) as fp: |
| 61 | with gzip.GzipFile(fileobj=fp, mode="wb") as gzfp: |
| 62 | for x in data: |
| 63 | if drop_builtin: |
| 64 | x = {k: v for k, v in x.items() if not k.startswith("_")} |
| 65 | gzfp.write((json.dumps(x) + "\n").encode("utf-8")) |
| 66 | else: |
| 67 | with open(filename, mode) as fp: |
| 68 | for x in data: |
| 69 | if drop_builtin: |
| 70 | x = {k: v for k, v in x.items() if not k.startswith("_")} |
| 71 | fp.write((json.dumps(x) + "\n").encode("utf-8")) |
| 72 | |
| 73 | |
| 74 | def stream_jsonl(filename: str) -> Iterable[Dict]: |
no outgoing calls
no test coverage detected