Differential D4183 Diff 14731 swh/deposit/cli/client.py

Changeset 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:
		vlorentzUnsubmitted Not Done Inline Actions doesn't need to be a string anymore vlorentz: doesn't need to be a string anymore
		ardumontAuthorUnsubmitted Done Inline Actions i thought as much but it failed without it... see one iteration before that last diff update... ardumont: i thought as much but it failed without it... see one iteration before that last diff update...
		vlorentzUnsubmitted Not Done Inline Actions see my other comment vlorentz: see my other comment
		ardumontAuthorUnsubmitted Done Inline Actions oh thanks, i did not notice it... I'll fix in another diff (i built on this one and now, lazyness hit me ;) ardumont: oh thanks, i did not notice it... I'll fix in another diff (i built on this one and now…
		ardumontAuthorUnsubmitted Done Inline Actions D4208 ardumont: 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:
		vlorentzUnsubmitted Not Done Inline Actions same 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)