I/O and Streaming
I/O and Streaming
The boring default for text is a file object you iterate: one line at a time, closed by a with block. io.StringIO is the same shape in memory, which is how you test a writer without touching disk.
Mental model
A text stream (TextIO) is something you can read, write, and iterate as lines. Real files from Path.open are streams. io.StringIO is a stream backed by a string. Functions that take TextIO work on both.
Streaming means you process a chunk (usually a line) and let it go. read() pulls the whole file into one str. That is fine for a 20-line ticket list. It is a habit that blows up on a month of logs.
tempfile.TemporaryDirectory still owns any real files in this chapter.
Worked examples
Case 1: StringIO as a fake report file
Save as report_buffer.py. The writer only knows it has a stream. getvalue() is the in-memory extra.
# report_buffer.py
from io import StringIO
def write_shift_report(buf: StringIO, tables: list[int]) -> None:
buf.write("shift report\n")
for n in tables:
buf.write(f"table {n} open\n")
def main() -> None:
buf = StringIO()
write_shift_report(buf, [3, 11, 12])
print(buf.getvalue(), end="")
if __name__ == "__main__":
main()Run:
uv run python report_buffer.pyOutput:
shift report
table 3 open
table 11 open
table 12 open
Case 2: Iterate a real file by line
Save as iter_tickets.py. for line in fh reads incrementally. split is enough for this format.
# iter_tickets.py
from pathlib import Path
from tempfile import TemporaryDirectory
def main() -> None:
with TemporaryDirectory() as raw:
path = Path(raw) / "tickets.txt"
path.write_text("7 12\n8 3\n9 11\n", encoding="utf-8")
with path.open(encoding="utf-8") as fh:
for line in fh:
ticket_id, table = line.split()
print(f"ticket {ticket_id} → table {table}")
if __name__ == "__main__":
main()Run:
uv run python iter_tickets.pyOutput:
ticket 7 → table 12
ticket 8 → table 3
ticket 9 → table 11
The with closes the file even if a later line raises.
Case 3: Accept any TextIO
Save as copy_tables.py. typing.TextIO is the type for a text stream. The function copies without caring whether the ends are files or StringIO.
# copy_tables.py
from io import StringIO
from typing import TextIO
def copy_open_tables(src: TextIO, dest: TextIO) -> int:
n = 0
for line in src:
dest.write(line)
n += 1
return n
def main() -> None:
src = StringIO("3\n11\n")
dest = StringIO()
count = copy_open_tables(src, dest)
print(count)
print(dest.getvalue(), end="")
if __name__ == "__main__":
main()Run:
uv run python copy_tables.pyOutput:
2
3
11
Case 4: Stream from a temp file into StringIO
Save as file_to_buffer.py. Same helper, real file on the left.
# file_to_buffer.py
from io import StringIO
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import TextIO
def copy_open_tables(src: TextIO, dest: TextIO) -> int:
n = 0
for line in src:
dest.write(line)
n += 1
return n
def main() -> None:
with TemporaryDirectory() as raw:
path = Path(raw) / "open.txt"
path.write_text("12\n4\n", encoding="utf-8")
dest = StringIO()
with path.open(encoding="utf-8") as src:
copy_open_tables(src, dest)
print(dest.getvalue(), end="")
if __name__ == "__main__":
main()Run:
uv run python file_to_buffer.pyOutput:
12
4
The trap
read() then splitlines() works until the file is large. This program slurps on purpose. Prefer the for line in fh loop from Case 2 when the file can grow.
Save as slurp_tickets.py:
# slurp_tickets.py
from pathlib import Path
from tempfile import TemporaryDirectory
def main() -> None:
with TemporaryDirectory() as raw:
path = Path(raw) / "tickets.txt"
path.write_text("7 12\n8 3\n", encoding="utf-8")
text = path.read_text(encoding="utf-8")
for line in text.splitlines():
ticket_id, table = line.split()
print(f"ticket {ticket_id} → table {table}")
if __name__ == "__main__":
main()Run:
uv run python slurp_tickets.pyOutput:
ticket 7 → table 12
ticket 8 → table 3
The output matches the streaming version. The cost does not. Keep read_text for small configs. Iterate logs and exports.
The boring rule
- Type writers as
TextIO(orStringIOwhen you needgetvalue). - Iterate lines for anything that might grow.
- Always
open(..., encoding="utf-8"). - Close with
with. Do not rely on the garbage collector. - Use
StringIOin tests; useTemporaryDirectorywhen you need a real path.
Try this
- In
report_buffer.py, add a final linetables: NwhereNislen(tables). - In
iter_tickets.py, skip blank lines (if not line.strip(): continue). - Change
copy_open_tablesso it uppercases each line as it copies.