# -*- coding: utf-8 -*-
#
# This file is part of DataCite.
#
# Copyright (C) 2015 CERN.
#
# DataCite is free software; you can redistribute it and/or modify it
# under the terms of the Revised BSD License; see LICENSE file for
# more details.
"""Python API client wrapper for the DataCite Metadata Store API.
API documentation is available on https://mds.datacite.org/static/apidoc.
"""
from __future__ import absolute_import, print_function
from .errors import DataCiteError
from .request import DataCiteRequest
[docs]class DataCiteMDSClient(object):
"""DataCite MDS API client wrapper."""
def __init__(self, username=None, password=None, url=None, prefix=None,
test_mode=False, api_ver="2", timeout=None):
"""Initialize the API client wrapper.
:param username: DataCite username.
:param password: DataCite password.
:param url: DataCite API base URL. Defaults to
https://mds.datacite.org/.
:param prefix: DOI prefix (or CFG_DATACITE_DOI_PREFIX). Defaults to
10.5072 (DataCite test prefix).
:param test_mode: Set to True to enable test mode. Defaults to False.
:param api_ver: DataCite API version. Currently has no effect.
Default to 2.
:param timeout: Connect and read timeout in seconds. Specify a tuple
(connect, read) to specify each timeout individually.
"""
self.username = username
self.password = password
self.prefix = prefix or '10.5072'
self.api_ver = api_ver # Currently not used
self.api_url = url or 'https://mds.datacite.org/'
if self.api_url[-1] != '/':
self.api_url = self.api_url + "/"
# If in test mode, set prefix to 10.5072, the default DataCite test
# prefix.
self.test_mode = test_mode
self.timeout = timeout
def __repr__(self):
"""Create string representation of object."""
return '<DataCiteMDSClient: {0}>'.format(self.username)
def _request_factory(self):
"""Create a new Request object."""
params = {}
if self.test_mode:
params['testMode'] = '1'
return DataCiteRequest(
base_url=self.api_url,
username=self.username,
password=self.password,
default_params=params,
timeout=self.timeout,
)
[docs] def doi_get(self, doi):
"""Get the URL where the resource pointed by the DOI is located.
:param doi: DOI name of the resource.
"""
r = self._request_factory()
r.get("doi/" + doi)
if r.code == 200:
return r.data
else:
raise DataCiteError.factory(r.code, r.data)
[docs] def doi_post(self, new_doi, location):
"""Mint new DOI.
:param new_doi: DOI name for the new resource.
:param location: URL where the resource is located.
:return: "CREATED" or "HANDLE_ALREADY_EXISTS".
"""
headers = {'Content-Type': 'text/plain;charset=UTF-8'}
# Use \r\n for HTTP client data.
body = "\r\n".join(["doi=%s" % new_doi, "url=%s" % location])
r = self._request_factory()
r.post("doi", body=body, headers=headers)
if r.code == 201:
return r.data
else:
raise DataCiteError.factory(r.code, r.data)
[docs] def metadata_post(self, metadata):
"""Set new metadata for an existing DOI.
Metadata should follow the DataCite Metadata Schema:
http://schema.datacite.org/
:param metadata: XML format of the metadata.
:return: "CREATED" or "HANDLE_ALREADY_EXISTS"
"""
headers = {'Content-Type': 'application/xml;charset=UTF-8', }
r = self._request_factory()
r.post("metadata", body=metadata, headers=headers)
if r.code == 201:
return r.data
else:
raise DataCiteError.factory(r.code, r.data)
[docs] def media_post(self, doi, media):
"""Add/update media type/urls pairs to a DOI.
Standard domain restrictions check will be performed.
:param media: Dictionary of (mime-type, URL) key/value pairs.
:return: "OK"
"""
headers = {'Content-Type': 'text/plain;charset=UTF-8'}
# Use \r\n for HTTP client data.
body = "\r\n".join(["%s=%s" % (k, v) for k, v in media.items()])
r = self._request_factory()
r.post("media/" + doi, body=body, headers=headers)
if r.code == 200:
return r.data
else:
raise DataCiteError.factory(r.code, r.data)