Differential D4431 Diff 15735 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

from __future__ import annotations		from __future__ import annotations

from datetime import datetime, timezone		from datetime import datetime, timezone
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		from typing import TYPE_CHECKING, Any, Collection, Dict, List, Optional

import click		import click

from swh.deposit.cli import deposit		from swh.deposit.cli import deposit

logger = logging.getLogger(__name__)		logger = logging.getLogger(__name__)


▲ Show 20 Lines • Show All 82 Lines • ▼ Show 20 Lines	codemetadata = {
{"codemeta:name": author_name} for author_name in authors		{"codemeta:name": author_name} for author_name in authors
],		],
},		},
}		}
logging.debug("Metadata dict to generate as xml: %s", codemetadata)		logging.debug("Metadata dict to generate as xml: %s", codemetadata)
return xmltodict.unparse(codemetadata, pretty=True)		return xmltodict.unparse(codemetadata, pretty=True)


def _client(url: str, username: str, password: str) -> PublicApiDepositClient:
"""Instantiate a client to access the deposit api server

Args:
url (str): Deposit api server
username (str): User
password (str): User's password

"""
from swh.deposit.client import PublicApiDepositClient

return PublicApiDepositClient(
{"url": url, "auth": {"username": username, "password": password},}
)


def _collection(client: PublicApiDepositClient) -> str:		def _collection(client: PublicApiDepositClient) -> str:
"""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(
		client,
username: str,		username: str,
password: str,
archive: Optional[str],		archive: Optional[str],
metadata: Optional[str],		metadata: Optional[str],
		collection: Optional[str],
archive_deposit: bool,		archive_deposit: bool,
metadata_deposit: bool,		metadata_deposit: bool,
collection: Optional[str],
slug: Optional[str],		slug: Optional[str],
partial: bool,		partial: bool,
deposit_id: Optional[int],		deposit_id: Optional[int],
swhid: Optional[str],		swhid: Optional[str],
replace: bool,		replace: bool,
url: str,		url: str,
name: Optional[str],		name: Optional[str],
authors: List[str],		authors: List[str],
Show All 26 Lines	Raises:
InputError explaining the user input related issue		InputError explaining the user input related issue
MaintenanceError explaining the api status		MaintenanceError explaining the api status

Returns:		Returns:
dict with the following keys:		dict with the following keys:

'archive': the software archive to deposit		'archive': the software archive to deposit
'username': username		'username': username
'password': associated password
'metadata': the metadata file to deposit		'metadata': the metadata file to deposit
'collection': the username's associated client		'collection: the user's collection under which to put the deposit
'slug': the slug or external id identifying the deposit to make		'slug': the slug or external id identifying the deposit to make
'partial': if the deposit is partial or not		'partial': if the deposit is partial or not
'client': instantiated class		'client': instantiated class
'url': deposit's server main entry point		'url': deposit's server main entry point
'deposit_type': deposit's type (binary, multipart, metadata)		'deposit_type': deposit's type (binary, multipart, metadata)
'deposit_id': optional deposit identifier		'deposit_id': optional deposit identifier
'swhid': optional deposit swhid		'swhid': optional deposit swhid

▲ Show 20 Lines • Show All 55 Lines • ▼ Show 20 Lines	if archive_deposit:
metadata = None		metadata = None

if not archive and not metadata and partial:		if not archive and not metadata and partial:
raise InputError(		raise InputError(
"Please provide an actionable command. See --help for more information"		"Please provide an actionable command. See --help for more information"
)		)

if replace and not deposit_id:		if replace and not deposit_id:
raise InputError("To update an existing deposit, you must provide its id")		raise InputError("To update an existing deposit, you must provide its id")
		ardumontUnsubmitted Not Done Inline Actions It's already there line 254 below. ardumont: It's already there line 254 below.
		douarddaAuthorUnsubmitted Done Inline Actions oops, thx douardda: oops, thx

client = _client(url, username, password)

if not collection:		if not collection:
collection = _collection(client)		collection = _collection(client)

return {		return {
"archive": archive,		"archive": archive,
"username": username,		"username": username,
"password": password,
"metadata": metadata,		"metadata": metadata,
"collection": collection,		"collection": collection,
"slug": slug,		"slug": slug,
"in_progress": partial,		"in_progress": partial,
"client": client,
"url": url,		"url": url,
"deposit_id": deposit_id,		"deposit_id": deposit_id,
"swhid": swhid,		"swhid": swhid,
"replace": replace,		"replace": replace,
}		}


def _subdict(d: Dict[str, Any], keys: Tuple[str, ...]) -> Dict[str, Any]:		def _subdict(d: Dict[str, Any], keys: Collection[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: Dict[str, Any]) -> Dict[str, Any]:
"""Delegate the actual deposit to the deposit client.

"""
logger.debug("Create deposit")

client = config["client"]
keys = ("collection", "archive", "metadata", "slug", "in_progress")
return client.deposit_create(**_subdict(config, keys))


def deposit_update(config: Dict[str, Any]) -> Dict[str, Any]:
"""Delegate the actual deposit to the deposit client.

"""
logger.debug("Update deposit")

client = config["client"]
keys = (
"collection",
"deposit_id",
"archive",
"metadata",
"slug",
"in_progress",
"replace",
"swhid",
)
return client.deposit_update(**_subdict(config, keys))


@deposit.command()		@deposit.command()
@click.option("--username", required=True, help="(Mandatory) User's name")		@click.option("--username", required=True, help="(Mandatory) User's name")
@click.option(		@click.option(
"--password", required=True, help="(Mandatory) User's associated password"		"--password", required=True, help="(Mandatory) User's associated password"
)		)
@click.option(		@click.option(
"--archive",		"--archive",
type=click.Path(exists=True),		type=click.Path(exists=True),
▲ Show 20 Lines • Show All 101 Lines • ▼ Show 20 Lines	):
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.

"""		"""
import tempfile		import tempfile

from swh.deposit.client import MaintenanceError		from swh.deposit.client import MaintenanceError, PublicApiDepositClient

url = _url(url)		url = _url(url)
config = {}

		client = PublicApiDepositClient(url=url, auth=(username, password))
with tempfile.TemporaryDirectory() as temp_dir:		with tempfile.TemporaryDirectory() as temp_dir:
try:		try:
logger.debug("Parsing cli options")		logger.debug("Parsing cli options")
config = client_command_parse_input(		config = client_command_parse_input(
		client,
username,		username,
password,
archive,		archive,
metadata,		metadata,
		collection,
archive_deposit,		archive_deposit,
metadata_deposit,		metadata_deposit,
collection,
slug,		slug,
partial,		partial,
deposit_id,		deposit_id,
swhid,		swhid,
replace,		replace,
url,		url,
name,		name,
author,		author,
temp_dir,		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"]		keys = ["archive", "collection", "in_progress", "metadata", "slug"]
		if config["deposit_id"]:
if deposit_id:		keys += ["deposit_id", "replace", "swhid"]
data = deposit_update(config)		data = client.deposit_update(**_subdict(config, keys))
else:		else:
data = deposit_create(config)		data = client.deposit_create(**_subdict(config, keys))

print_result(data, 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, "
Show All 12 Lines	@click.option(
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 status(ctx, url, username, password, deposit_id, output_format):		def status(ctx, url, username, password, deposit_id, output_format):
"""Deposit's status		"""Deposit's status

"""		"""
from swh.deposit.client import MaintenanceError		from swh.deposit.client import MaintenanceError, PublicApiDepositClient

url = _url(url)		url = _url(url)
logger.debug("Status deposit")		logger.debug("Status deposit")
try:		try:
client = _client(url, username, password)		client = PublicApiDepositClient(url=url, auth=(username, password))
collection = _collection(client)		collection = _collection(client)
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)

Show All 20 Lines