You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
145 lines
5.9 KiB
145 lines
5.9 KiB
from __future__ import annotations
|
|
|
|
import base64
|
|
import ssl
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import anyio
|
|
|
|
|
|
DAV_NAMESPACE = {"d": "DAV:"}
|
|
|
|
|
|
class WebDAVStorageAdapter:
|
|
def __init__(self, config: dict[str, Any]) -> None:
|
|
self.config = config
|
|
self.base_url = config["endpoint_url"].rstrip("/")
|
|
self.root_path = self._normalize_relative_path(config.get("root_path") or "iron")
|
|
self._ssl_context = None if config.get("verify_ssl", True) else ssl._create_unverified_context()
|
|
|
|
async def check(self) -> dict[str, str]:
|
|
def _check() -> dict[str, str]:
|
|
self._ensure_directory(self.root_path)
|
|
return {"status": "healthy", "detail": self.root_path}
|
|
|
|
return await anyio.to_thread.run_sync(_check)
|
|
|
|
async def put_file(self, source_path: str, storage_key: str) -> dict[str, str | None]:
|
|
def _put() -> dict[str, str | None]:
|
|
source = Path(source_path)
|
|
if not source.exists():
|
|
raise FileNotFoundError(source_path)
|
|
normalized_key = self._normalize_relative_path(storage_key)
|
|
segments = normalized_key.split("/")
|
|
if len(segments) > 1:
|
|
self._ensure_directory(f"{self.root_path}/{'/'.join(segments[:-1])}")
|
|
request = urllib.request.Request(
|
|
self._build_url(f"{self.root_path}/{normalized_key}"),
|
|
data=source.read_bytes(),
|
|
method="PUT",
|
|
headers=self._auth_headers(),
|
|
)
|
|
with urllib.request.urlopen(request, timeout=60, context=self._ssl_context) as response:
|
|
etag = response.headers.get("ETag")
|
|
return {"etag": etag, "storage_key": normalized_key}
|
|
|
|
return await anyio.to_thread.run_sync(_put)
|
|
|
|
async def download_file(self, storage_key: str, destination_path: Path) -> Path:
|
|
def _download() -> Path:
|
|
normalized_key = self._normalize_relative_path(storage_key)
|
|
request = urllib.request.Request(
|
|
self._build_url(f"{self.root_path}/{normalized_key}"),
|
|
method="GET",
|
|
headers=self._auth_headers(),
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=60, context=self._ssl_context) as response:
|
|
body = response.read()
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 404:
|
|
raise FileNotFoundError(storage_key) from exc
|
|
raise
|
|
destination_path.parent.mkdir(parents=True, exist_ok=True)
|
|
destination_path.write_bytes(body)
|
|
return destination_path
|
|
|
|
return await anyio.to_thread.run_sync(_download)
|
|
|
|
async def object_exists(self, storage_key: str) -> bool:
|
|
def _exists() -> bool:
|
|
normalized_key = self._normalize_relative_path(storage_key)
|
|
request = urllib.request.Request(
|
|
self._build_url(f"{self.root_path}/{normalized_key}"),
|
|
method="HEAD",
|
|
headers=self._auth_headers(),
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=20, context=self._ssl_context):
|
|
return True
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 404:
|
|
return False
|
|
if exc.code in {403, 405}:
|
|
return self._path_exists(f"{self.root_path}/{normalized_key}")
|
|
raise
|
|
|
|
return await anyio.to_thread.run_sync(_exists)
|
|
|
|
def _ensure_directory(self, relative_path: str) -> None:
|
|
current = ""
|
|
for segment in self._normalize_relative_path(relative_path).split("/"):
|
|
current = f"{current}/{segment}".strip("/")
|
|
if self._path_exists(current):
|
|
continue
|
|
request = urllib.request.Request(
|
|
self._build_url(current),
|
|
method="MKCOL",
|
|
headers=self._auth_headers(),
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=20, context=self._ssl_context):
|
|
pass
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code not in {201, 405}:
|
|
raise RuntimeError(f"WebDAV MKCOL failed with {exc.code} for {current}") from exc
|
|
|
|
def _path_exists(self, relative_path: str) -> bool:
|
|
request = urllib.request.Request(
|
|
self._build_url(relative_path),
|
|
method="PROPFIND",
|
|
headers={**self._auth_headers(), "Depth": "0"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=20, context=self._ssl_context) as response:
|
|
body = response.read()
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 404:
|
|
return False
|
|
raise RuntimeError(f"WebDAV PROPFIND failed with {exc.code} for {relative_path}") from exc
|
|
if not body:
|
|
return True
|
|
root = ET.fromstring(body)
|
|
return root.find("d:response", DAV_NAMESPACE) is not None
|
|
|
|
def _build_url(self, relative_path: str) -> str:
|
|
safe_path = "/".join(urllib.parse.quote(part, safe="") for part in relative_path.strip("/").split("/") if part)
|
|
return f"{self.base_url}/{safe_path}"
|
|
|
|
def _auth_headers(self) -> dict[str, str]:
|
|
username = str(self.config.get("username") or "")
|
|
password = str(self.config.get("password") or "")
|
|
token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
|
|
return {"Authorization": f"Basic {token}"}
|
|
|
|
@staticmethod
|
|
def _normalize_relative_path(value: str) -> str:
|
|
normalized = "/".join(part for part in value.strip().strip("/").split("/") if part)
|
|
if not normalized:
|
|
raise ValueError("path must not be empty")
|
|
return normalized
|
|
|