123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
import os
|
|
from contextlib import contextmanager
|
|
from datetime import timedelta
|
|
from functools import cache
|
|
from typing import Any, AnyStr, TypeAlias
|
|
from collections.abc import Generator
|
|
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
|
|
from .config import REQUESTS_RETRY, REQUESTS_TIMEOUT, Config
|
|
|
|
PathLike: TypeAlias = os.PathLike | AnyStr
|
|
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
|
|
|
|
|
|
def requests_cache_disabled() -> Any:
|
|
import requests_cache
|
|
return requests_cache.disabled()
|
|
|
|
|
|
@cache
|
|
def get_requests_session(nocache: bool = False) -> requests.Session:
|
|
adapter = HTTPAdapter(max_retries=REQUESTS_RETRY)
|
|
if nocache:
|
|
with requests_cache_disabled():
|
|
http = requests.Session()
|
|
else:
|
|
http = requests.Session()
|
|
http.mount("https://", adapter)
|
|
http.mount("http://", adapter)
|
|
return http
|
|
|
|
|
|
@contextmanager
|
|
def install_requests_cache() -> Generator:
|
|
# This adds basic etag based caching, to avoid hitting API rate limiting
|
|
|
|
import requests_cache
|
|
from requests_cache.backends.sqlite import SQLiteCache
|
|
|
|
# Monkey patch globally, so pygithub uses it as well.
|
|
# Only do re-validation with etag/date etc and ignore the cache-control headers that
|
|
# github sends by default with 60 seconds.
|
|
cache_dir = os.path.join(os.getcwd(), '.autobuild_cache')
|
|
os.makedirs(cache_dir, exist_ok=True)
|
|
cache_file = f'http_cache_{requests_cache.__version__}.sqlite'
|
|
# delete other versions
|
|
for f in os.listdir(cache_dir):
|
|
if f.startswith('http_cache') and f != cache_file:
|
|
os.remove(os.path.join(cache_dir, f))
|
|
requests_cache.install_cache(
|
|
always_revalidate=True,
|
|
cache_control=False,
|
|
expire_after=requests_cache.EXPIRE_IMMEDIATELY,
|
|
backend=SQLiteCache(os.path.join(cache_dir, cache_file)))
|
|
|
|
# Call this once, so it gets cached from the main thread and can be used in a thread pool
|
|
get_requests_session(nocache=True)
|
|
|
|
try:
|
|
yield
|
|
finally:
|
|
# Delete old cache entries, so this doesn't grow indefinitely
|
|
cache = requests_cache.get_cache()
|
|
assert cache is not None
|
|
cache.delete(older_than=timedelta(hours=3))
|
|
|
|
# un-monkey-patch again
|
|
requests_cache.uninstall_cache()
|
|
|
|
|
|
@contextmanager
|
|
def gha_group(title: str) -> Generator:
|
|
print(f'\n::group::{title}')
|
|
try:
|
|
yield
|
|
finally:
|
|
print('::endgroup::')
|
|
|
|
|
|
def queue_website_update() -> None:
|
|
session = get_requests_session()
|
|
r = session.post('https://packages.msys2.org/api/trigger_update', timeout=REQUESTS_TIMEOUT)
|
|
try:
|
|
# it's not worth stopping the build if this fails, so just log it
|
|
r.raise_for_status()
|
|
except requests.RequestException as e:
|
|
print(e)
|
|
|
|
|
|
def parse_optional_deps(optional_deps: str) -> dict[str, list[str]]:
|
|
res: dict[str, list[str]] = {}
|
|
optional_deps = optional_deps.replace(" ", "")
|
|
if not optional_deps:
|
|
return res
|
|
for entry in optional_deps.split(","):
|
|
assert ":" in entry
|
|
first, second = entry.split(":", 2)
|
|
res.setdefault(first, []).append(second)
|
|
return res
|
|
|
|
|
|
def apply_optional_deps(optional_deps: str) -> None:
|
|
for dep, ignored in parse_optional_deps(optional_deps).items():
|
|
Config.OPTIONAL_DEPS.setdefault(dep, []).extend(ignored)
|
|
|
|
|
|
def ask_yes_no(prompt, default_no: bool = True):
|
|
"""Ask a yes/no question via input() and return their answer."""
|
|
|
|
if default_no:
|
|
prompt += " [y/N] "
|
|
else:
|
|
prompt += " [Y/n] "
|
|
|
|
user_input = input(prompt).strip().lower()
|
|
|
|
if not user_input:
|
|
return False if default_no else True
|
|
else:
|
|
return user_input == 'y'
|