Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/cli/client.py
# Copyright (C) 2017-2020 The Software Heritage developers | # Copyright (C) 2017-2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import logging | import logging | ||||
# WARNING: do not import unnecessary things here to keep cli startup time under | # WARNING: do not import unnecessary things here to keep cli startup time under | ||||
# control | # control | ||||
import os | import os | ||||
import sys | import sys | ||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple | |||||
import click | import click | ||||
from swh.deposit.cli import deposit | from swh.deposit.cli import deposit | ||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
if TYPE_CHECKING: | |||||
from swh.deposit.client import PublicApiDepositClient | |||||
class InputError(ValueError): | class InputError(ValueError): | ||||
"""Input script error | """Input script error | ||||
""" | """ | ||||
pass | pass | ||||
def generate_slug(): | def generate_slug() -> str: | ||||
"""Generate a slug (sample purposes). | """Generate a slug (sample purposes). | ||||
""" | """ | ||||
import uuid | import uuid | ||||
return str(uuid.uuid4()) | return str(uuid.uuid4()) | ||||
def _url(url): | def _url(url: str) -> str: | ||||
"""Force the /1 api version at the end of the url (avoiding confusing | """Force the /1 api version at the end of the url (avoiding confusing | ||||
issues without it). | issues without it). | ||||
Args: | Args: | ||||
url (str): api url used by cli users | url (str): api url used by cli users | ||||
Returns: | Returns: | ||||
Top level api url to actually request | Top level api url to actually request | ||||
""" | """ | ||||
if not url.endswith("/1"): | if not url.endswith("/1"): | ||||
url = "%s/1" % url | url = "%s/1" % url | ||||
return url | return url | ||||
def generate_metadata_file(name, external_id, authors, temp_dir): | def generate_metadata_file( | ||||
name: str, external_id: str, authors: List[str], temp_dir: str | |||||
) -> str: | |||||
"""Generate a temporary metadata file with the minimum required metadata | """Generate a temporary metadata file with the minimum required metadata | ||||
This generates a xml file in a temporary location and returns the | This generates a xml file in a temporary location and returns the | ||||
path to that file. | path to that file. | ||||
This is up to the client of that function to clean up the | This is up to the client of that function to clean up the | ||||
temporary file. | temporary file. | ||||
Args: | Args: | ||||
name (str): Software's name | name: Software's name | ||||
external_id (str): External identifier (slug) or generated one | external_id: External identifier (slug) or generated one | ||||
authors (List[str]): List of author names | authors: List of author names | ||||
Returns: | Returns: | ||||
Filepath to the metadata generated file | Filepath to the metadata generated file | ||||
""" | """ | ||||
import xmltodict | import xmltodict | ||||
path = os.path.join(temp_dir, "metadata.xml") | path = os.path.join(temp_dir, "metadata.xml") | ||||
Show All 14 Lines | ) -> str: | ||||
logging.debug("Metadata dict to generate as xml: %s", codemetadata) | logging.debug("Metadata dict to generate as xml: %s", codemetadata) | ||||
s = xmltodict.unparse(codemetadata, pretty=True) | s = xmltodict.unparse(codemetadata, pretty=True) | ||||
logging.debug("Metadata dict as xml generated: %s", s) | logging.debug("Metadata dict as xml generated: %s", s) | ||||
with open(path, "w") as fp: | with open(path, "w") as fp: | ||||
fp.write(s) | fp.write(s) | ||||
return path | return path | ||||
def _client(url, username, password): | def _client(url: str, username: str, password: str) -> "PublicApiDepositClient": | ||||
vlorentz: doesn't need to be a string anymore | |||||
Done Inline Actionsi thought as much but it failed without it... ardumont: i thought as much but it failed without it...
see one iteration before that last diff update... | |||||
Not Done Inline Actionssee my other comment vlorentz: see my other comment | |||||
Done Inline Actionsoh thanks, i did not notice it... ardumont: oh thanks, i did not notice it...
I'll fix in another diff (i built on this one and now… | |||||
Done Inline Actionsardumont: D4208 | |||||
"""Instantiate a client to access the deposit api server | """Instantiate a client to access the deposit api server | ||||
Args: | Args: | ||||
url (str): Deposit api server | url (str): Deposit api server | ||||
username (str): User | username (str): User | ||||
password (str): User's password | password (str): User's password | ||||
""" | """ | ||||
from swh.deposit.client import PublicApiDepositClient | from swh.deposit.client import PublicApiDepositClient | ||||
client = PublicApiDepositClient( | return PublicApiDepositClient( | ||||
{"url": url, "auth": {"username": username, "password": password},} | {"url": url, "auth": {"username": username, "password": password},} | ||||
) | ) | ||||
return client | |||||
def _collection(client): | def _collection(client: "PublicApiDepositClient") -> str: | ||||
Not Done Inline Actionssame vlorentz: same | |||||
"""Retrieve the client's collection | """Retrieve the client's collection | ||||
""" | """ | ||||
# retrieve user's collection | # retrieve user's collection | ||||
sd_content = client.service_document() | sd_content = client.service_document() | ||||
if "error" in sd_content: | if "error" in sd_content: | ||||
raise InputError("Service document retrieval: %s" % (sd_content["error"],)) | raise InputError("Service document retrieval: %s" % (sd_content["error"],)) | ||||
collection = sd_content["service"]["workspace"]["collection"]["sword:name"] | collection = sd_content["service"]["workspace"]["collection"]["sword:name"] | ||||
return collection | return collection | ||||
def client_command_parse_input( | def client_command_parse_input( | ||||
username, | username: str, | ||||
password, | password: str, | ||||
archive, | archive: Optional[str], | ||||
metadata, | metadata: Optional[str], | ||||
archive_deposit, | archive_deposit: bool, | ||||
metadata_deposit, | metadata_deposit: bool, | ||||
collection, | collection: Optional[str], | ||||
slug, | slug: Optional[str], | ||||
partial, | partial: bool, | ||||
deposit_id, | deposit_id: Optional[int], | ||||
replace, | replace: bool, | ||||
url, | url: str, | ||||
name, | name: Optional[str], | ||||
authors, | authors: List[str], | ||||
temp_dir, | temp_dir: str, | ||||
): | ) -> Dict[str, Any]: | ||||
"""Parse the client subcommand options and make sure the combination | """Parse the client subcommand options and make sure the combination | ||||
is acceptable*. If not, an InputError exception is raised | is acceptable*. If not, an InputError exception is raised | ||||
explaining the issue. | explaining the issue. | ||||
By acceptable, we mean: | By acceptable, we mean: | ||||
- A multipart deposit (create or update) requires: | - A multipart deposit (create or update) requires: | ||||
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Lines | return { | ||||
"in_progress": partial, | "in_progress": partial, | ||||
"client": client, | "client": client, | ||||
"url": url, | "url": url, | ||||
"deposit_id": deposit_id, | "deposit_id": deposit_id, | ||||
"replace": replace, | "replace": replace, | ||||
} | } | ||||
def _subdict(d, keys): | def _subdict(d: Dict[str, Any], keys: Tuple[str, ...]) -> Dict[str, Any]: | ||||
"return a dict from d with only given keys" | "return a dict from d with only given keys" | ||||
return {k: v for k, v in d.items() if k in keys} | return {k: v for k, v in d.items() if k in keys} | ||||
def deposit_create(config, logger): | def deposit_create(config: Dict[str, Any]) -> Dict[str, Any]: | ||||
"""Delegate the actual deposit to the deposit client. | """Delegate the actual deposit to the deposit client. | ||||
""" | """ | ||||
logger.debug("Create deposit") | logger.debug("Create deposit") | ||||
client = config["client"] | client = config["client"] | ||||
keys = ("collection", "archive", "metadata", "slug", "in_progress") | keys = ("collection", "archive", "metadata", "slug", "in_progress") | ||||
return client.deposit_create(**_subdict(config, keys)) | return client.deposit_create(**_subdict(config, keys)) | ||||
def deposit_update(config, logger): | def deposit_update(config: Dict[str, Any]) -> Dict[str, Any]: | ||||
"""Delegate the actual deposit to the deposit client. | """Delegate the actual deposit to the deposit client. | ||||
""" | """ | ||||
logger.debug("Update deposit") | logger.debug("Update deposit") | ||||
client = config["client"] | client = config["client"] | ||||
keys = ( | keys = ( | ||||
"collection", | "collection", | ||||
▲ Show 20 Lines • Show All 86 Lines • ▼ Show 20 Lines | @click.option( | ||||
"output_format", | "output_format", | ||||
default="logging", | default="logging", | ||||
type=click.Choice(["logging", "yaml", "json"]), | type=click.Choice(["logging", "yaml", "json"]), | ||||
help="Output format results.", | help="Output format results.", | ||||
) | ) | ||||
@click.pass_context | @click.pass_context | ||||
def upload( | def upload( | ||||
ctx, | ctx, | ||||
username, | username: str, | ||||
password, | password: str, | ||||
archive=None, | archive: Optional[str] = None, | ||||
metadata=None, | metadata: Optional[str] = None, | ||||
archive_deposit=False, | archive_deposit: bool = False, | ||||
metadata_deposit=False, | metadata_deposit: bool = False, | ||||
collection=None, | collection: Optional[str] = None, | ||||
slug=None, | slug: Optional[str] = None, | ||||
partial=False, | partial: bool = False, | ||||
deposit_id=None, | deposit_id: Optional[int] = None, | ||||
replace=False, | replace: bool = False, | ||||
url="https://deposit.softwareheritage.org", | url: str = "https://deposit.softwareheritage.org", | ||||
verbose=False, | verbose: bool = False, | ||||
name=None, | name: Optional[str] = None, | ||||
author=None, | author: List[str] = [], | ||||
output_format=None, | output_format: Optional[str] = None, | ||||
): | ): | ||||
"""Software Heritage Public Deposit Client | """Software Heritage Public Deposit Client | ||||
Create/Update deposit through the command line. | Create/Update deposit through the command line. | ||||
More documentation can be found at | More documentation can be found at | ||||
https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html. | https://docs.softwareheritage.org/devel/swh-deposit/getting-started.html. | ||||
Show All 28 Lines | with tempfile.TemporaryDirectory() as temp_dir: | ||||
except InputError as e: | except InputError as e: | ||||
logger.error("Problem during parsing options: %s", e) | logger.error("Problem during parsing options: %s", e) | ||||
sys.exit(1) | sys.exit(1) | ||||
except MaintenanceError as e: | except MaintenanceError as e: | ||||
logger.error(e) | logger.error(e) | ||||
sys.exit(1) | sys.exit(1) | ||||
if verbose: | if verbose: | ||||
logger.info("Parsed configuration: %s" % (config,)) | logger.info("Parsed configuration: %s", config) | ||||
deposit_id = config["deposit_id"] | deposit_id = config["deposit_id"] | ||||
if deposit_id: | if deposit_id: | ||||
r = deposit_update(config, logger) | data = deposit_update(config) | ||||
else: | else: | ||||
r = deposit_create(config, logger) | data = deposit_create(config) | ||||
print_result(r, output_format) | print_result(data, output_format) | ||||
@deposit.command() | @deposit.command() | ||||
@click.option( | @click.option( | ||||
"--url", | "--url", | ||||
default="https://deposit.softwareheritage.org", | default="https://deposit.softwareheritage.org", | ||||
help="(Optional) Deposit server api endpoint. By default, " | help="(Optional) Deposit server api endpoint. By default, " | ||||
"https://deposit.softwareheritage.org/1", | "https://deposit.softwareheritage.org/1", | ||||
Show All 31 Lines | except MaintenanceError as e: | ||||
sys.exit(1) | sys.exit(1) | ||||
print_result( | print_result( | ||||
client.deposit_status(collection=collection, deposit_id=deposit_id), | client.deposit_status(collection=collection, deposit_id=deposit_id), | ||||
output_format, | output_format, | ||||
) | ) | ||||
def print_result(data, output_format): | def print_result(data: Dict[str, Any], output_format: Optional[str]) -> None: | ||||
"""Display the result data into a dedicated output format. | |||||
""" | |||||
import json | import json | ||||
import yaml | import yaml | ||||
if output_format == "json": | if output_format == "json": | ||||
click.echo(json.dumps(data)) | click.echo(json.dumps(data)) | ||||
elif output_format == "yaml": | elif output_format == "yaml": | ||||
click.echo(yaml.dump(data)) | click.echo(yaml.dump(data)) | ||||
else: | else: | ||||
logger.info(data) | logger.info(data) |
doesn't need to be a string anymore