feat(cli): add private NanoShare sync client
Build and Push Docker Container / build-and-push (push) Successful in 1m3s
Build and Push Docker Container / build-and-push (push) Successful in 1m3s
- Add a separate installable NanoShare CLI package under cli/. - Implement browser login with local callback and refresh-token config. - Add upload, list, download, sync, and watch CLI commands. - Add private owner-only file download endpoint for CLI downloads. - Add CLI auth endpoints for browser login and token refresh. - Return file_id from ServiceLink uploads for reliable sync state. - Exclude the CLI package from NanoShare container builds. - Include tests for private downloads and sync state updates.
This commit is contained in:
@@ -0,0 +1,68 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import tomllib
|
||||
except ImportError: # pragma: no cover
|
||||
tomllib = None
|
||||
|
||||
from .config import DEFAULT_CONFIG
|
||||
from .http_client import NanoShareClient
|
||||
|
||||
def parse_registry_env(value: str) -> dict[str, str]:
|
||||
registry: dict[str, str] = {}
|
||||
for pair in value.split(','):
|
||||
pair = pair.strip()
|
||||
if not pair:
|
||||
continue
|
||||
name, _, url = pair.partition('=')
|
||||
if name.strip() and url.strip():
|
||||
registry[name.strip()] = url.strip()
|
||||
return registry
|
||||
|
||||
def load_config(path: str | os.PathLike | None) -> dict:
|
||||
if not path or tomllib is None:
|
||||
return {}
|
||||
config_path = Path(path).expanduser()
|
||||
if not config_path.is_file():
|
||||
return {}
|
||||
with config_path.open('rb') as handle:
|
||||
return tomllib.load(handle)
|
||||
|
||||
def resolve(args) -> tuple[dict[str, str], str, dict[str, str | None]]:
|
||||
config_path = args.config or os.getenv('NANOSHARE_CONFIG') or DEFAULT_CONFIG
|
||||
config = load_config(config_path)
|
||||
|
||||
registry: dict[str, str] = dict(config.get('nodes', {}))
|
||||
env_registry = os.getenv('NANOSHARE_REGISTRY')
|
||||
if env_registry:
|
||||
registry.update(parse_registry_env(env_registry))
|
||||
for item in (args.url or []):
|
||||
name, _, url = item.partition('=')
|
||||
if not url:
|
||||
raise SystemExit(f'--url expects NODE=URL, got: {item!r}')
|
||||
registry[name.strip()] = url.strip()
|
||||
|
||||
source = args.source or os.getenv('NANOSHARE_SOURCE') or config.get('source') or 'nanoshare-cli'
|
||||
auth = {
|
||||
'token': args.token or os.getenv('NANOSHARE_TOKEN') or config.get('token'),
|
||||
'refresh_token': getattr(args, 'refresh_token', None) or os.getenv('NANOSHARE_REFRESH_TOKEN') or config.get('refresh_token'),
|
||||
'token_url': getattr(args, 'token_url', None) or os.getenv('NANOSHARE_TOKEN_URL') or config.get('token_url'),
|
||||
}
|
||||
return registry, source, auth
|
||||
|
||||
def make_client(args) -> NanoShareClient:
|
||||
registry, source, auth = resolve(args)
|
||||
if not registry:
|
||||
raise SystemExit('no NanoShare node configured (run `nanoshare login BASE_URL`, pass --url, or set NANOSHARE_REGISTRY).')
|
||||
return NanoShareClient(
|
||||
source=source,
|
||||
registry=registry,
|
||||
token=auth.get('token'),
|
||||
refresh_token=auth.get('refresh_token'),
|
||||
token_url=auth.get('token_url'),
|
||||
timeout=getattr(args, 'timeout', 30.0),
|
||||
)
|
||||
Reference in New Issue
Block a user