diff --git a/src/streaming/extract_stream.py b/src/streaming/extract_stream.py index 29bb8cf..b04c43f 100644 --- a/src/streaming/extract_stream.py +++ b/src/streaming/extract_stream.py @@ -6,7 +6,7 @@ import io import lzma import tarfile import tempfile -from collections.abc import Iterable, Iterator +from collections.abc import Iterable, Iterator, Callable from dataclasses import dataclass from pathlib import Path from urllib.parse import urlparse @@ -21,10 +21,12 @@ class ExtractConfig: def _infer_name_from_url(url: str) -> str: - """Extract filename from URL path using string utilities (not Path for cross-platform compatibility).""" + """Extract filename from URL path using string utilities + (not Path for cross-platform compatibility).""" try: path = urlparse(url).path - # Use string operations instead of Path() to avoid Windows/Unix path separator issues + # Use string operations instead of Path() + # to avoid Windows/Unix path separator issues return path.rstrip('/').rsplit('/', 1)[-1] if path else "stream" except Exception: return "stream" @@ -77,7 +79,7 @@ def iter_tar_member_bytes( tar_name: str, byte_iter: Iterable[bytes], *, - member_predicate: callable | None = None, + member_predicate: Callable | None = None, ) -> Iterator[tuple[str, Iterator[bytes]]]: """Stream members from a tar/tar.gz without writing to disk. @@ -92,8 +94,7 @@ def iter_tar_member_bytes( elif tar_name.endswith(".tar"): mode = "r:" - tf = tarfile.open(fileobj=raw, mode=mode) - try: + with tarfile.open(fileobj=raw, mode=mode) as tf: for member in tf: if not member.isfile(): continue @@ -103,8 +104,6 @@ def iter_tar_member_bytes( if f is None: continue yield member.name, _iter_fileobj_bytes(f) - finally: - tf.close() def iter_text_lines( @@ -115,10 +114,13 @@ def iter_text_lines( ) -> Iterator[str]: """Convert an iterable of bytes into an iterator of decoded text lines.""" # Use incremental decoder and splitlines while preserving newlines. - decoder = io.TextIOWrapper(io.BufferedReader(_IterableBytesIO(byte_iter)), encoding=encoding, errors=errors) + decoder = io.TextIOWrapper( + io.BufferedReader(_IterableBytesIO(byte_iter)), + encoding=encoding, + errors=errors + ) try: - for line in decoder: - yield line + yield from decoder finally: decoder.detach() @@ -130,14 +132,15 @@ def spool_zip_to_tempfile( ) -> Path: """Spool a ZIP stream to a temporary file (ZIP requires seekable input).""" cfg = cfg or ExtractConfig() - tmp = tempfile.NamedTemporaryFile(delete=False, dir=str(cfg.zip_spool_dir) if cfg.zip_spool_dir else None, suffix=cfg.zip_spool_suffix) - try: + with tempfile.NamedTemporaryFile( + delete=False, + dir=str(cfg.zip_spool_dir) if cfg.zip_spool_dir else None, + suffix=cfg.zip_spool_suffix + ) as tmp: for chunk in byte_iter: tmp.write(chunk) tmp.flush() return Path(tmp.name) - finally: - tmp.close() class _IterableBytesIO(io.RawIOBase): diff --git a/src/streaming/http_stream.py b/src/streaming/http_stream.py index 72bf5cb..81276d0 100644 --- a/src/streaming/http_stream.py +++ b/src/streaming/http_stream.py @@ -56,7 +56,11 @@ def stream_http_bytes( if chunk: yield chunk return - except (httpx.TimeoutException, httpx.NetworkError, httpx.RemoteProtocolError, httpx.HTTPStatusError) as e: + except ( + httpx.TimeoutException, + httpx.NetworkError, + httpx.RemoteProtocolError, + httpx.HTTPStatusError) as e: # Retry on network/timeouts and 5xx. For 4xx (except 429), fail fast. if isinstance(e, httpx.HTTPStatusError): status = e.response.status_code