mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-11-10 12:28:27 +00:00
fix(types): type download_url's return by spool_to_disk (#1649)
The Python Typechecks job fails on main since #1632, which is why #1644 shows red too. `download_url` returns a `BytesIO`, or a `SpooledTemporaryFile` when `spool_to_disk` is set, and both were typed as `BinaryIO`. That has no `getbuffer()`, which four direct-download callers use. Two overloads on `spool_to_disk` give each caller the type it actually gets, so nothing changes at runtime. Thanks @29rdx for the spooling work, this only tightens the types around it. basedpyright (source and tests), ruff and vulture are clean, and the tests for the download, direct download and libgen code pass. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> <!-- coderabbit-ai-generated:v1 --> ## Summary by CodeRabbit * **Developer Experience** * Updated download API type information to distinguish between in-memory buffers and disk-spooled buffers. The return type now reflects the selected spooling option, making the API’s possible results clearer to developers and improving type checking when working with downloaded content. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
1 parent
9d8e7eaad2
commit
fa42f357d2
1 file changed
+48
-5
@@ -5,7 +5,7 @@ import time
|
||||
from http import HTTPStatus
|
||||
from io import BytesIO
|
||||
from tempfile import SpooledTemporaryFile
|
||||
from typing import TYPE_CHECKING, BinaryIO, NoReturn
|
||||
from typing import TYPE_CHECKING, Literal, NoReturn, overload
|
||||
from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse
|
||||
|
||||
import requests
|
||||
@@ -874,6 +874,46 @@ def html_get_page(
|
||||
)
|
||||
|
||||
|
||||
@overload
|
||||
def download_url(
|
||||
link: str,
|
||||
size: str = "",
|
||||
progress_callback: Callable[[float], None] | None = None,
|
||||
cancel_flag: Event | None = None,
|
||||
_selector: network.AAMirrorSelector | None = None,
|
||||
status_callback: Callable[[str, str | None], None] | None = None,
|
||||
referer: str | None = None,
|
||||
spool_to_disk: Literal[False] = False,
|
||||
) -> BytesIO | None: ...
|
||||
|
||||
|
||||
@overload
|
||||
def download_url(
|
||||
link: str,
|
||||
size: str = "",
|
||||
progress_callback: Callable[[float], None] | None = None,
|
||||
cancel_flag: Event | None = None,
|
||||
_selector: network.AAMirrorSelector | None = None,
|
||||
status_callback: Callable[[str, str | None], None] | None = None,
|
||||
referer: str | None = None,
|
||||
*,
|
||||
spool_to_disk: Literal[True],
|
||||
) -> SpooledTemporaryFile[bytes] | None: ...
|
||||
|
||||
|
||||
@overload
|
||||
def download_url(
|
||||
link: str,
|
||||
size: str,
|
||||
progress_callback: Callable[[float], None] | None,
|
||||
cancel_flag: Event | None,
|
||||
_selector: network.AAMirrorSelector | None,
|
||||
status_callback: Callable[[str, str | None], None] | None,
|
||||
referer: str | None,
|
||||
spool_to_disk: Literal[True],
|
||||
) -> SpooledTemporaryFile[bytes] | None: ...
|
||||
|
||||
|
||||
def download_url(
|
||||
link: str,
|
||||
size: str = "",
|
||||
@@ -883,8 +923,11 @@ def download_url(
|
||||
status_callback: Callable[[str, str | None], None] | None = None,
|
||||
referer: str | None = None,
|
||||
spool_to_disk: bool = False,
|
||||
) -> BinaryIO | None:
|
||||
"""Download content from URL with automatic retry and resume support."""
|
||||
) -> BytesIO | SpooledTemporaryFile[bytes] | None:
|
||||
"""Download content from URL with automatic retry and resume support.
|
||||
|
||||
Returns a ``BytesIO``, or a ``SpooledTemporaryFile`` when ``spool_to_disk`` is set.
|
||||
"""
|
||||
selector = _selector or network.AAMirrorSelector()
|
||||
current_url = selector.rewrite(link)
|
||||
try:
|
||||
@@ -1071,13 +1114,13 @@ def _is_configured_zlib_host(hostname: str | None) -> bool:
|
||||
|
||||
def _try_resume(
|
||||
url: str,
|
||||
buffer: BinaryIO,
|
||||
buffer: BytesIO | SpooledTemporaryFile[bytes],
|
||||
start_byte: int,
|
||||
total_size: float,
|
||||
progress_callback: Callable[[float], None] | None,
|
||||
cancel_flag: Event | None,
|
||||
base_headers: dict | None = None,
|
||||
) -> BinaryIO | None:
|
||||
) -> BytesIO | SpooledTemporaryFile[bytes] | None:
|
||||
"""Try to resume an interrupted download."""
|
||||
for attempt in range(MAX_RESUME_ATTEMPTS):
|
||||
logger.info(
|
||||
|
||||
Reference in new issue
Block a user