"""Zugriff auf den Server der Software-Bibliothek. Nur die beiden Routen, die aus dem Netzwerk erreichbar sind: Katalog lesen und Datei holen. Die Verwaltung bleibt der Weboberflaeche am Server vorbehalten. Bewusst nur mit der Standardbibliothek — jede Abhaengigkeit mehr muss beim Bau der exe mitgeschleppt werden, und fuer zwei GET-Aufrufe lohnt das nicht. """ from __future__ import annotations import json import urllib.error import urllib.request from pathlib import Path from typing import Any, Callable ZEITGRENZE_KATALOG_S = 15 ZEITGRENZE_DOWNLOAD_S = 600 class BibliothekNichtErreichbar(Exception): """Der Server antwortet nicht oder liefert etwas Unerwartetes.""" def _lesbarer_fehler(adresse: str, fehler: Exception) -> BibliothekNichtErreichbar: if isinstance(fehler, urllib.error.HTTPError): return BibliothekNichtErreichbar(f"{adresse} antwortet mit HTTP {fehler.code}.") if isinstance(fehler, urllib.error.URLError): return BibliothekNichtErreichbar(f"{adresse} ist nicht erreichbar: {fehler.reason}") return BibliothekNichtErreichbar(f"{adresse}: {fehler}") def katalog(server_adresse: str) -> list[dict[str, Any]]: """Holt alle Katalogeintraege.""" adresse = f"{server_adresse.rstrip('/')}/api/software" try: with urllib.request.urlopen(adresse, timeout=ZEITGRENZE_KATALOG_S) as antwort: gelesen = json.loads(antwort.read().decode("utf-8")) except Exception as fehler: # noqa: BLE001 - jede Ursache wird zur selben Meldung raise _lesbarer_fehler(server_adresse, fehler) from fehler if not isinstance(gelesen, list): raise BibliothekNichtErreichbar(f"{server_adresse} liefert keinen Katalog.") return gelesen def datei_holen( server_adresse: str, software_id: int, ziel: Path, fortschritt: Callable[[int, int], None] | None = None, ) -> Path: """Laedt die bereitgestellte Datei nach `ziel`. Geschrieben wird zuerst nach `.teil` und erst nach vollstaendigem Empfang umbenannt. Ein Abbruch mitten im Download darf keine halbe Datei hinterlassen, die spaeter fuer ein fertiges Programm gehalten wird. """ adresse = f"{server_adresse.rstrip('/')}/api/software/{software_id}/download" unfertig = ziel.with_name(ziel.name + ".teil") ziel.parent.mkdir(parents=True, exist_ok=True) try: with urllib.request.urlopen(adresse, timeout=ZEITGRENZE_DOWNLOAD_S) as antwort: gesamt = int(antwort.headers.get("Content-Length") or 0) geladen = 0 with unfertig.open("wb") as datei: while True: brocken = antwort.read(256 * 1024) if not brocken: break datei.write(brocken) geladen += len(brocken) if fortschritt: fortschritt(geladen, gesamt) except Exception as fehler: # noqa: BLE001 unfertig.unlink(missing_ok=True) raise _lesbarer_fehler(server_adresse, fehler) from fehler if unfertig.stat().st_size == 0: unfertig.unlink(missing_ok=True) raise BibliothekNichtErreichbar("Der Server hat eine leere Datei geliefert.") ziel.unlink(missing_ok=True) unfertig.replace(ziel) return ziel