mirror of
https://github.com/ankitects/anki.git
synced 2025-09-19 14:32:22 -04:00
b9251290ca
This adds Python 3.9 and 3.10 typing syntax to files that import attributions from __future___. Python 3.9 should be able to cope with the 3.10 syntax, but Python 3.8 will no longer work. On Windows/Mac, install the latest Python 3.9 version from python.org. There are currently no orjson wheels for Python 3.10 on Windows/Mac, which will break the build unless you have Rust installed separately. On Linux, modern distros should have Python 3.9 available already. If you're on an older distro, you'll need to build Python from source first.
88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
# Copyright: Ankitects Pty Ltd and contributors
|
|
# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
|
|
|
|
"""
|
|
Wrapper for requests that adds a callback for tracking upload/download progress.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
from typing import Any, Callable
|
|
|
|
import requests
|
|
from requests import Response
|
|
|
|
HTTP_BUF_SIZE = 64 * 1024
|
|
|
|
ProgressCallback = Callable[[int, int], None]
|
|
|
|
|
|
class HttpClient:
|
|
|
|
verify = True
|
|
timeout = 60
|
|
# args are (upload_bytes_in_chunk, download_bytes_in_chunk)
|
|
progress_hook: ProgressCallback | None = None
|
|
|
|
def __init__(self, progress_hook: ProgressCallback | None = None) -> None:
|
|
self.progress_hook = progress_hook
|
|
self.session = requests.Session()
|
|
|
|
def __enter__(self) -> HttpClient:
|
|
return self
|
|
|
|
def __exit__(self, *args: Any) -> None:
|
|
self.close()
|
|
|
|
def close(self) -> None:
|
|
if self.session:
|
|
self.session.close()
|
|
self.session = None
|
|
|
|
def __del__(self) -> None:
|
|
self.close()
|
|
|
|
def post(self, url: str, data: bytes, headers: dict[str, str] | None) -> Response:
|
|
headers["User-Agent"] = self._agentName()
|
|
return self.session.post(
|
|
url,
|
|
data=data,
|
|
headers=headers,
|
|
stream=True,
|
|
timeout=self.timeout,
|
|
verify=self.verify,
|
|
) # pytype: disable=wrong-arg-types
|
|
|
|
def get(self, url: str, headers: dict[str, str] = None) -> Response:
|
|
if headers is None:
|
|
headers = {}
|
|
headers["User-Agent"] = self._agentName()
|
|
return self.session.get(
|
|
url, stream=True, headers=headers, timeout=self.timeout, verify=self.verify
|
|
)
|
|
|
|
def streamContent(self, resp: Response) -> bytes:
|
|
resp.raise_for_status()
|
|
|
|
buf = io.BytesIO()
|
|
for chunk in resp.iter_content(chunk_size=HTTP_BUF_SIZE):
|
|
if self.progress_hook:
|
|
self.progress_hook(0, len(chunk))
|
|
buf.write(chunk)
|
|
return buf.getvalue()
|
|
|
|
def _agentName(self) -> str:
|
|
from anki.buildinfo import version
|
|
|
|
return f"Anki {version}"
|
|
|
|
|
|
# allow user to accept invalid certs in work/school settings
|
|
if os.environ.get("ANKI_NOVERIFYSSL"):
|
|
HttpClient.verify = False
|
|
|
|
import warnings
|
|
|
|
warnings.filterwarnings("ignore")
|