*** Wartungsfenster jeden ersten Mittwoch vormittag im Monat ***

Skip to content
Snippets Groups Projects
Commit 8ab488c0 authored by Moser, Maximilian's avatar Moser, Maximilian
Browse files

Implement `auto.download()` of all files for a container

parent efba3713
Branches
Tags
1 merge request!5v0.2.0
......@@ -5,7 +5,7 @@
"""Helpers for interacting automatically with various repository types."""
from typing import Optional, Tuple
from typing import List, Optional, Tuple
from urllib.parse import urlparse
import pandas as pd
......@@ -74,38 +74,61 @@ def suggest_repository(url: str) -> Optional[BaseWrapper]:
if service_cls is None:
return None
return service_cls(url)
return service_cls(f"{scheme}://{host}")
def download(url: str, all: bool = False, interactive: bool = True) -> Optional[str]:
def _download(
service: BaseWrapper, container_id: str, files: List[str] | str, interactive: bool
) -> List[str] | str:
"""Download one or more files."""
results = []
already_auth, single_file = False, False
if isinstance(files, (str, int)):
files, single_file = [files], True
for file in files:
try:
dl = service.download(container_id, file)
results.append(dl)
except Exception as e:
if interactive and not already_auth:
service.authenticate_interactive()
already_auth = True
dl = service.download(container_id, file)
results.append(dl)
else:
raise e
return results[0] if single_file else results
def download(
url: str, all: bool = False, interactive: bool = True
) -> List[str] | str | None:
"""Download file automatically based on the URL."""
if (service := suggest_repository(url)) is None:
return None
# fish out the container & file from the URL
cid, fid = service.url_to_parts(url)
cid, fid, fids = *service.url_to_parts(url), []
if cid is None and fid is None:
cid, fid = service.url_to_parts(_follow_redirects(url))
# if we couldn't determine a file, ask the user in interactive mode
# if we couldn't determine a file from the URL...
if fid is None:
if all:
# TODO get a list of all files
pass
# we either take all of them
fids = list(service.list_files(cid))
elif interactive:
# TODO ask the user which file to download
# or we ask the user (TODO)
pass
else:
# or we give up
return None
try:
return service.download(cid, fid)
except Exception:
# if the download didn't work, we can try again with authentication
if interactive:
service.authenticate_interactive()
return service.download(cid, fid)
return _download(service, cid, fid or fids, interactive)
finally:
service.clear_auth()
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment