Skip to content
Snippets Groups Projects
Commit a8d91687 authored by Romin's avatar Romin :eye:
Browse files

Merge branch 'development' into 'master'

Development

See merge request coscine/docs/public/coscine-python-client!2
parents 7fe63e74 77f46183
No related branches found
Tags v0.5.0
1 merge request!2Development
Showing
with 1558 additions and 773 deletions
__pycache__/
test/
config.json
token.txt
\ No newline at end of file
#
# This ci script builds the python package and pushes a release
# to the python package index (pypi).
# It is triggered upon the creation of a tag with name vX.X.X where
# X is a number in [0, 9].
# For the deployment to pypi the token 'PYPI_TOKEN' is required
# to be present in the repository.
#
stages:
- release
......
# Changelog
## [WIP] 0.5.0 - 2021-07-07
## 0.5.0 - 2021-08-16
```diff
+ Revamp of the entire module
+ Type Hinting
+ Native S3-integration with boto3
+ Member management
+ Metadata Configuration in ResourceForm
+ Metadata preset configuration in ResourceForm
+ Various code examples
- DOCStrings
- breaking changes to previous versions
```
## 0.4.0 - 2021-06-02
......
......@@ -2,20 +2,30 @@
![Coscine](data/logo.png)
Coscine is an integration platform for research data hosted at RWTH Aachen University.
Visit the [Landing Page](https://coscine.pages.rwth-aachen.de/docs/public/pages) for
more information.
[Coscine](https://coscine.rwth-aachen.de/) is an integration platform for
research data hosted at RWTH Aachen University. For more information visit
the [Landing Page](https://coscine.de/).
### About
## About
This python library provides a simple API client for Coscine.
This python3.x library provides a simple interface to Coscine. You can use it
to integrate Coscine in your python applications and easily:
- create or delete projects and resources
- invite and manage project members
- upload and download files, resources or even whole projects
- manage your metadata
### Documentation
Even if you plan on using the API directly in a programming language of your choice,
you might want to take a look at this module as it provides you with a handy
debug mode which logs all requests and responses made in the communication with coscine.
This makes it simple to understand how certain workflows should be implemented or why
something in your implementation might not work.
Documentation and Installation instructions can be found [here](https://git.rwth-aachen.de/coscine/docs/public/coscine-api-python-client/-/wikis/home).
Additionally DOCstrings have been used to document the source code.
## Documentation
### Contact
Documentation and installation instructions can be found [here](https://git.rwth-aachen.de/coscine/docs/public/coscine-api-python-client/-/wikis/home).
## Contact
To report bugs, request features or resolve questions open an issue inside
of the current git repository.
......@@ -26,7 +36,7 @@ open a merge request.
For general help or consulting contact RWTH Aachen
IT ServiceDesk at <servicedesk@rwth-aachen.de>.
### License
## License
This project is Open Source Software and licensed
under the terms of the MIT License.
......@@ -50,4 +60,4 @@ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
```
\ No newline at end of file
```
import setuptools
from src.coscine.about import __version__, __author__
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setuptools.setup(
name = "coscine",
version = "0.4.1",
description = "Coscine Python3 Client",
version = __version__,
description = "Coscine Python Client",
long_description = long_description,
long_description_content_type = "text/markdown",
author = "RWTH Aachen University",
author = __author__,
author_email = "coscine@itc.rwth-aachen.de",
license = "MIT License",
packages = setuptools.find_packages(where="src"),
keywords = [
"Coscine", "RWTH Aachen", "RDM", "Research Data Management"
"Coscine", "RWTH Aachen", "Research Data Management"
],
install_requires = [
"requests",
"requests-toolbelt",
"tqdm",
"colorama"
"colorama",
"boto3"
],
url = "https://git.rwth-aachen.de/coscine/docs/public/coscine-api-python-client",
url = "https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client",
project_urls = {
"Bug Tracker": "https://git.rwth-aachen.de/coscine/docs/public/coscine-api-python-client/-/issues"
"Issues": "https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client/-/issues",
"Wiki": "https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client/-/wikis/home"
},
classifiers = [
"Programming Language :: Python :: 3",
......@@ -33,5 +36,5 @@ setuptools.setup(
"Intended Audience :: Developers"
],
package_dir = {"": "src"},
python_requires = ">=3.6"
)
python_requires = ">=3.7"
)
\ No newline at end of file
###############################################################################
# Coscine Python3 Client
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/wiki/-/tree/master/
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# This python wrapper implements a client for the Coscine API.
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.rwth-aachen.de for more information.
# Visit https://coscine.de for more information.
###############################################################################
import os
from collections.abc import MutableMapping
from collections import OrderedDict
from .exceptions import *
from .form import Form
###############################################################################
class MetadataForm(MutableMapping):
class MetadataForm(Form):
"""
Coscine Metadata Input Form
Can be used to generate and manipulate json-ld formatted metadata
"""
###############################################################################
def __init__(self, handle, project, resource):
self.store = {}
self._keys = {}
self.vocabulary = {}
self.handle = handle
self.profile = handle.get_application_profile(resource, parse=True)
for entry in self.profile["graph"]:
if "class" in entry:
uri = entry["class"]
name = entry["name"][self.handle.lang]
data = handle.get_instance(project, uri)
lang = self.handle.lang
vocabulary = {}
if lang not in data:
lang = "en"
for entry in data[lang]:
vocabulary[entry["name"]] = entry["value"]
self.vocabulary[name] = vocabulary
self.reset()
###############################################################################
def __getitem__(self, key):
return self.store[key]
###############################################################################
def __setitem__(self, key, value):
if key not in self._keys:
raise KeyError(key)
elif self.is_controlled(key):
vocabulary = self.get_vocabulary(key)
if type(value) is list:
self.store[key] = []
for val in value:
if val in vocabulary:
self.store[key].append(vocabulary[val])
else:
raise VocabularyError(val)
else:
if value in vocabulary:
self.store[key] = vocabulary[value]
else:
raise VocabularyError(value)
else:
self.store[key] = value
###############################################################################
def __delitem__(self, key):
del self.store[key]
###############################################################################
def __iter__(self):
return iter(self.store)
###############################################################################
def __len__(self):
return len(self.store)
def __init__(self, profile: dict, lang: str, entries: list, vocabulary: dict):
super().__init__("Metadata Form", lang, entries, vocabulary)
self.profile = profile
###############################################################################
def __repr__(self):
entries = []
for key in self._keys:
R = " "
C = " "
value = ""
if self.is_required(key):
R = "R"
if self.is_controlled(key):
C = "C"
if key in self.store:
value = " = %s" % self.store[key]
text = " [%s%s] %s%s" % (R, C, key, value)
entries.append(text)
format = \
"_______________________________\n\n" \
" Coscine Metadata Form\n" \
"_______________________________\n\n" \
" [R: Required] [C: Controlled]\n" \
"-------------------------------\n" \
+ "\n".join(entries) + \
"\n_______________________________\n"
return format
###############################################################################
def is_required(self, key):
"""
Determines wether a key is required
Parameters
-----------
key : str
Returns
--------
bool
"""
def parse(self, data: dict):
info = self._keys[key]
if "minCount" in info and info["minCount"] > 0:
return True
else:
return False
###############################################################################
def is_controlled(self, key):
"""
Determines wether a key is controlled by a vocabulary
Parameters
-----------
key : str
Returns
--------
bool
"""
if key in self.vocabulary:
return True
else:
return False
###############################################################################
def get_vocabulary(self, key):
"""
Returns the vocabulary associated with a controlled key
Parameters
-----------
key : str
Returns
--------
dict
"""
if self.is_controlled(key):
return self._keys[key]["vocabulary"]
else:
msg = "Key [%s] is not controlled by a vocabulary!" % key
raise CoscineException(msg)
###############################################################################
def keys(self):
"""
Enumerates the keys of the input form
"""
keys = []
for key in self._keys:
keys.append(key)
return keys
###############################################################################
def reset(self):
"""
Resets the input form to default
Parses JSON-LD metadata into a Metadata Input Form
"""
self.store.clear()
for entry in self.profile["graph"]:
name = entry["name"][self.handle.lang]
self._keys[name] = entry
if self.is_controlled(name):
self._keys[name]["vocabulary"] = self.vocabulary[name]
# Sort the keys according to their application profile order
tuples = sorted(self._keys.items(), key = lambda x: x[1]["order"])
self._keys = OrderedDict(tuples)
k = list(data.keys())[0]
for path in data[k]:
for entry in self.profile["graph"]:
if path == entry["path"]:
key = entry["name"][self._lang]
value = data[k][path][0]["value"]
if self.is_controlled(key):
voc = self.get_vocabulary(key)
keys = list(voc.keys())
values = list(voc.values())
index = values.index(value)
value = keys[index]
self.store[key] = value
break
###############################################################################
def generate(self):
"""
Generates JSON-LD formatted metadata representation
Raises
------
RequirementError
When one or more required have not been set
Returns
--------
JSON-LD formatted metadata
"""
def generate(self) -> dict:
metadata = {}
RDFTYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
# Set application profile type used for the metadata
# Set application profile type used by metadata
metadata[RDFTYPE] = [{
"type": "uri",
"value": self.profile["id"]
......@@ -247,16 +62,24 @@ class MetadataForm(MutableMapping):
# Set metadata fields
for key in self._keys:
entry = self._keys[key]
field = entry["field"]
if key not in self.store:
if self.is_required(key):
missing.append(key)
else:
field = self._keys[key]
path = field["path"]
path = entry["path"]
value = self.store[key]
if entry["flags"] & MetadataForm.LIST:
if type(value) is not list and type(value) is not tuple:
raise ValueError("Expected iterable value for key %s" % key)
if self.is_controlled(key):
voc = self.get_vocabulary(key)
value = voc[value]
metadata[path] = [{
"value": self.store[key],
"datatype": field["datatype"],
"type": field["type"]
"value": value,
"datatype": entry["datatype"],
"type": entry["type"]
}]
# Check missing field list
......@@ -266,17 +89,3 @@ class MetadataForm(MutableMapping):
return metadata
###############################################################################
def parse(self, data):
"""
Parses JSON-LD metadata into a Metadata Input Form
"""
for path in data:
for entry in self.profile["graph"]:
if path in entry["path"]:
self.store[entry["name"][self.handle.lang]] = data[path][0]["value"]
break
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
#
# The MetadataPreset form allows for the configuration of default
# values in an application profile.
#
from .form import Form
###############################################################################
class MetadataPresetForm(Form):
def __init__(self, profile: dict, lang: str, entries: list, vocabulary: dict):
super().__init__("Metadata Preset", lang, entries, vocabulary)
self.profile = profile
###############################################################################
def enable(self, key: str, enable: bool = True):
self._keys[key]["enabled"] = enable
###############################################################################
def lock(self, key: str, lock: bool = True):
self._keys[key]["locked"] = lock
###############################################################################
def generate(self) -> dict:
metadata = {}
# Set metadata fields
for key in self._keys:
field = self._keys[key]
path = field["path"]
if key not in self.store:
value = None
metadata[path] = {
"https://purl.org/coscine/defaultValue": [],
"https://purl.org/coscine/invisible": [{
"type": "literal",
"value": "0"
}]
}
continue
else:
value = self.store[key]
if self.is_controlled(key):
voc = self.get_vocabulary(key)
value = voc[value]
metadata[path] = {
"https://purl.org/coscine/defaultValue": [{
"type": field["type"],
"value": value
}],
"https://purl.org/coscine/invisible": [{
"type": "literal",
"value": "0"
}]
}
return metadata
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python3 Client
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/wiki/-/tree/master/
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# This python wrapper implements a client for the Coscine API.
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.rwth-aachen.de for more information.
# Visit https://coscine.de for more information.
###############################################################################
from collections.abc import MutableMapping
from .form import Form
from .exceptions import *
from .FormFlags import FormFlags
###############################################################################
KEYS = [
{
KEYS = [{
"name": {
"de": "Projektname",
"en": "Project Name"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "ProjectName"
},
{
},{
"name": {
"de": "Anzeigename",
"en": "Display Name"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "DisplayName"
},
{
},{
"name": {
"de": "Projektbeschreibung",
"en": "Project Description"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "Description"
},
{
},{
"name": {
"de": "Principal Investigators",
"en": "Principal Investigators"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "PrincipleInvestigators"
},
{
},{
"name": {
"de": "Projektstart",
"en": "Project Start"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "StartDate"
},
{
......@@ -64,278 +57,143 @@ KEYS = [
"de": "Projektende",
"en": "Project End"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "EndDate"
},
{
},{
"name": {
"de": "Disziplin",
"en": "Discipline"
},
"flags": FormFlags.REQUIRED | FormFlags.CONTROLLED,
"flags": Form.REQUIRED | Form.CONTROLLED | Form.LIST,
"field": "Discipline"
},
{
},{
"name": {
"de": "Teilnehmende Organisation",
"en": "Participating Organizations"
},
"flags": FormFlags.REQUIRED | FormFlags.CONTROLLED,
"flags": Form.REQUIRED | Form.CONTROLLED | Form.LIST,
"field": "Organization"
},
{
},{
"name": {
"de": "Projektschlagwörter",
"en": "Project Keywords"
},
"flags": FormFlags.NONE,
"flags": Form.NONE,
"field": "Keywords"
},
{
"name": {
"de": "Sichtbarkeit",
"en": "Visibility"
},
"flags": FormFlags.REQUIRED | FormFlags.CONTROLLED,
"field": "Visibility"
},
{
},{
"name": {
"de": "Grant ID",
"en": "Grant ID"
},
"flags": FormFlags.NONE,
"flags": Form.NONE,
"field": "GrantId"
},
{
},{
"name": {
"de": "Features",
"en": "Features"
},
"flags": FormFlags.CONTROLLED,
"flags": Form.CONTROLLED,
"field": "Features"
}
]
###############################################################################
}]
class ProjectForm(MutableMapping):
"""
Coscine Input Form for creating and editing projects
"""
MAP = {
"description": "Description",
"displayName": "DisplayName",
"startDate": "StartDate",
"endDate": "EndDate",
"keywords": "Keywords",
"projectName": "ProjectName",
"principleInvestigators": "PrincipleInvestigators",
"grantId": "GrantId",
"disciplines": "Discipline",
"organizations": "Organization"
}
###############################################################################
def __init__(self, handle, parent = None, data = None):
self.store = {}
self._keys = {}
self.handle = handle
disciplines = handle.get_disciplines()
organizations = handle.get_organizations()
visibility = handle.get_visibility()
features = handle.get_features()
self.vocabulary = {
"Discipline": disciplines,
"Organization": organizations,
"Visibility": visibility,
"Features": features
}
self.parent = parent
self.reset()
class ProjectForm(Form):
###############################################################################
def __getitem__(self, key):
return self.store[key]
def __init__(self, lang, vocabulary, parent):
super().__init__("Project Form", lang, KEYS, vocabulary)
self.parent = parent
###############################################################################
def __setitem__(self, key, value):
if key not in self._keys:
raise KeyError(key)
elif self.is_controlled(key):
vocabulary = self.get_vocabulary(key)
if type(value) is list:
self.store[key] = []
for val in value:
if val in vocabulary:
self.store[key].append(vocabulary[val])
else:
raise VocabularyError(val)
else:
if value in vocabulary:
self.store[key] = vocabulary[value]
def parse(self, data):
def _parse_disciplines(self, data):
LSTR = {"en": "displayNameEn", "de": "displayNameDe"}[self._lang]
disciplines = []
for discipline in data:
disciplines.append(discipline[LSTR])
return disciplines
def _parse_organizations(self, data):
organizations = []
for organization in data:
vocabulary = self._vocabulary["Organization"]
for entry in vocabulary.values():
if entry["url"] == organization["url"]:
organizations.append(entry["displayName"])
return organizations
for key in data:
if key in MAP and data[key] != "":
entry = self.entry(MAP[key])
if entry is None:
continue
name = entry["name"][self._lang]
flags = entry["flags"]
value = data[key]
if flags & Form.CONTROLLED or flags & Form.LIST:
if key == "disciplines":
self[name] = _parse_disciplines(self, value)
elif key == "organizations":
self[name] = _parse_organizations(self, value)
elif key == "visibility":
self[name] = value["displayName"]
else:
raise VocabularyError(value)
else:
self.store[key] = value
###############################################################################
def __delitem__(self, key):
del self.store[key]
###############################################################################
def __iter__(self):
return iter(self.store)
###############################################################################
def __len__(self):
return len(self.store)
###############################################################################
def is_required(self, key):
"""
Determines wether a key is required
Parameters
-----------
key : str
Returns
--------
bool
"""
if self._keys[key]["flags"] & FormFlags.REQUIRED:
return True
else:
return False
###############################################################################
def is_controlled(self, key):
"""
Determines wether a key is controlled by a vocabulary
Parameters
-----------
key : str
Returns
--------
bool
"""
if self._keys[key]["flags"] & FormFlags.CONTROLLED:
return True
else:
return False
###############################################################################
def get_vocabulary(self, key):
"""
Returns the vocabulary associated with a controlled key
Parameters
-----------
key : str
Returns
--------
dict
"""
if self.is_controlled(key):
return self._keys[key]["vocabulary"]
else:
msg = "Key [%s] is not controlled by a vocabulary!" % key
raise CoscineException(msg)
###############################################################################
def __repr__(self):
entries = []
for key in self._keys:
R = " "
C = " "
value = ""
if self.is_required(key):
R = "R"
if self.is_controlled(key):
C = "C"
if key in self.store:
value = " = %s" % self.store[key]
text = " [%s%s] %s%s" % (R, C, key, value)
entries.append(text)
format = \
"_______________________________\n\n" \
" Coscine Project Form\n" \
"_______________________________\n\n" \
" [R: Required] [C: Controlled]\n" \
"-------------------------------\n" \
+ "\n".join(entries) + \
"\n_______________________________\n"
return format
###############################################################################
def keys(self):
"""
Enumerates the keys of the input form
"""
keys = []
for key in self._keys:
keys.append(key)
return keys
###############################################################################
def reset(self):
"""
Resets the input form to default
"""
self.store.clear()
for key in KEYS:
self._keys[key["name"][self.handle.lang]] = key
if key["flags"] & FormFlags.CONTROLLED:
key["vocabulary"] = self.vocabulary[key["field"]]
self[name] = value
###############################################################################
def generate(self):
"""
Generates JSON-LD formatted representation of project data
Raises
-------
RequirementError
When one or more required fields have not been set
Returns
--------
JSON-LD formatted project data
"""
def final_value(key, value):
if self.is_controlled(key):
return self._vocabulary[field][value]
else:
return value
data = {}
missing = []
for key in self._keys:
value = self._keys[key]
entry = self._keys[key]
field = entry["field"]
if key not in self.store:
if self.is_required(key):
missing.append(key)
else:
data[value["field"]] = self.store[key]
value = self.store[key]
if entry["flags"] & ProjectForm.LIST:
if type(value) is not list and type(value) is not tuple:
raise ValueError("Expected iterable value for key %s" % key)
data[field] = []
for v in value:
data[field].append(final_value(key, v))
else:
data[field] = final_value(key, value)
if missing:
raise RequirementError(missing)
if self.parent:
data["ParentId"] = self.parent["id"]
data["ParentId"] = self.parent.id
data["Visibility"] = {
"displayName": "Project Members",
"id": "8ab9c883-eb0d-4402-aaad-2e4007badce6"
}
return data
......
###############################################################################
# Coscine Python3 Client
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/wiki/-/tree/master/
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# This python wrapper implements a client for the Coscine API.
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.rwth-aachen.de for more information.
# Visit https://coscine.de for more information.
###############################################################################
from collections.abc import MutableMapping
from .exceptions import *
from .FormFlags import FormFlags
from .form import Form
###############################################################################
KEYS = [
{
KEYS = [{
"name": {
"de": "Ressourcentyp",
"en": "Resource Type"
},
"flags": FormFlags.REQUIRED | FormFlags.CONTROLLED,
"flags": Form.REQUIRED | Form.CONTROLLED,
"field": "type"
},
{
},{ # This is REQUIRED for rds and s3, but not for Linked Data!
"name": {
"de": "Ressourcengröße",
"en": "Resource Size"
},
"flags": FormFlags.NONE,
"flags": Form.SPECIAL,
"field": "resourceTypeOption"
},
{
},{
"name": {
"de": "Ressourcenname",
"en": "Resource Name"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "ResourceName"
},
{
},{
"name": {
"de": "Anzeigename",
"en": "Display Name"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "DisplayName"
},
{
},{
"name": {
"de": "Ressourcenbeschreibung",
"en": "Resource Description"
},
"flags": FormFlags.REQUIRED,
"flags": Form.REQUIRED,
"field": "Description"
},
{
},{
"name": {
"de": "Disziplin",
"en": "Discipline"
},
"flags": FormFlags.REQUIRED | FormFlags.CONTROLLED,
"flags": Form.REQUIRED | Form.CONTROLLED | Form.LIST,
"field": "Disciplines"
},
{
},{
"name": {
"de": "Ressourcenschlagwörter",
"en": "Resource Keywords"
},
"flags": FormFlags.NONE,
"flags": Form.NONE,
"field": "Keywords"
},
{
"name": {
"de": "Sichtbarkeit",
"en": "Visibility"
},
"flags": FormFlags.REQUIRED | FormFlags.CONTROLLED,
"field": "Visibility"
},
{
},{
"name": {
"de": "Lizenz",
"en": "License"
},
"flags": FormFlags.CONTROLLED,
"flags": Form.CONTROLLED,
"field": "License"
},
{
},{
"name": {
"de": "Verwendungsrechte",
"en": "Usage Rights"
},
"flags": FormFlags.NONE,
"flags": Form.NONE,
"field": "UsageRights"
},
{
},{
"name": {
"de": "Applikationsprofile",
"en": "Application Profiles"
"en": "Application Profile"
},
"flags": FormFlags.REQUIRED | FormFlags.CONTROLLED,
"flags": Form.REQUIRED | Form.CONTROLLED,
"field": "applicationProfile"
}
]
}]
MAP = {
"description": "Description",
"displayName": "DisplayName",
"resourceName": "ResourceName",
"keywords": "Keywords",
"disciplines": "Disciplines",
"license": "License",
"resourceTypeOption": "resourceTypeOption",
"applicationProfile": "applicationProfile",
"type": "type",
"usageRights": "UsageRights"
}
###############################################################################
class ResourceForm(MutableMapping):
class ResourceForm(Form):
"""
Coscine Input Form for creating and editing resources
......@@ -119,187 +111,46 @@ class ResourceForm(MutableMapping):
###############################################################################
def __init__(self, handle):
self.store = {}
self._keys = {}
self.handle = handle
resource_types = handle.get_resource_types()
profiles = handle.get_profiles()
visibility = handle.get_visibility()
licenses = handle.get_licenses()
disciplines = handle.get_disciplines()
self.vocabulary = {
"type": resource_types,
"applicationProfile": profiles,
"Visibility": visibility,
"License": licenses,
"Disciplines": disciplines
}
self.reset()
###############################################################################
def __getitem__(self, key):
return self.store[key]
###############################################################################
def __setitem__(self, key, value):
if key not in self._keys:
raise KeyError(key)
elif self.is_controlled(key):
vocabulary = self.get_vocabulary(key)
if type(value) is list:
self.store[key] = []
for val in value:
if val in vocabulary:
self.store[key].append(vocabulary[val])
else:
raise VocabularyError(val)
else:
if value in vocabulary:
self.store[key] = vocabulary[value]
def __init__(self, lang: str, vocabulary: dict):
super().__init__("Resource Form", lang, KEYS, vocabulary)
###############################################################################
def parse(self, data: dict):
def _parse_disciplines(self, data):
LSTR = {"en": "displayNameEn", "de": "displayNameDe"}[self._lang]
disciplines = []
for discipline in data:
disciplines.append(discipline[LSTR])
return disciplines
for key in data:
if key in MAP and data[key] != "":
entry = self.entry(MAP[key])
if entry is None:
continue
name = entry["name"][self._lang]
flags = entry["flags"]
value = data[key]
if flags & Form.CONTROLLED or flags & Form.LIST or flags & Form.SPECIAL:
if key == "disciplines":
self[name] = _parse_disciplines(self, value)
elif key == "resourceTypeOption":
self[name] = value["Size"]
elif key == "applicationProfile":
values = list(self._vocabulary["applicationProfile"].values())
keys = list(self._vocabulary["applicationProfile"].keys())
position = values.index(value)
self[name] = keys[position]
elif key == "visibility" or key == "license" or key == "type":
self[name] = value["displayName"]
else:
raise VocabularyError(value)
else:
self.store[key] = value
###############################################################################
def __delitem__(self, key):
del self.store[key]
###############################################################################
def __iter__(self):
return iter(self.store)
###############################################################################
def __len__(self):
return len(self.store)
###############################################################################
def __repr__(self):
entries = []
for key in self._keys:
R = " "
C = " "
value = ""
if self.is_required(key):
R = "R"
if self.is_controlled(key):
C = "C"
if key in self.store:
value = " = %s" % self.store[key]
text = " [%s%s] %s%s" % (R, C, key, value)
entries.append(text)
format = \
"_______________________________\n\n" \
" Coscine Resource Form\n" \
"_______________________________\n\n" \
" [R: Required] [C: Controlled]\n" \
"-------------------------------\n" \
+ "\n".join(entries) + \
"\n_______________________________\n"
return format
###############################################################################
def is_required(self, key):
"""
Determines wether a key is required
Parameters
-----------
key : str
Returns
--------
bool
"""
if self._keys[key]["flags"] & FormFlags.REQUIRED:
return True
else:
return False
###############################################################################
def is_controlled(self, key):
"""
Determines wether a key is controlled by a vocabulary
Parameters
-----------
key : str
Returns
--------
bool
"""
if self._keys[key]["flags"] & FormFlags.CONTROLLED:
return True
else:
return False
###############################################################################
def get_vocabulary(self, key):
"""
Returns the vocabulary associated with a controlled key
Parameters
-----------
key : str
Returns
--------
dict
"""
if self.is_controlled(key):
return self._keys[key]["vocabulary"]
else:
msg = "Key [%s] is not controlled by a vocabulary!" % key
raise CoscineException(msg)
###############################################################################
def keys(self):
"""
Enumerates the keys of the input form
"""
keys = []
for key in self._keys:
keys.append(key)
return keys
###############################################################################
def reset(self):
"""
Resets the input form to default
"""
self.store.clear()
for key in KEYS:
self._keys[key["name"][self.handle.lang]] = key
if key["flags"] & FormFlags.CONTROLLED:
key["vocabulary"] = self.vocabulary[key["field"]]
self[name] = value
###############################################################################
def generate(self):
def generate(self) -> dict:
"""
Generates JSON-LD formatted representation of resource data
......@@ -314,15 +165,30 @@ class ResourceForm(MutableMapping):
JSON-LD formatted resource data
"""
def final_value(key, value):
if self.is_controlled(key):
return self._vocabulary[field][value]
else:
return value
data = {}
missing = []
for key in self._keys:
value = self._keys[key]
entry = self._keys[key]
field = entry["field"]
if key not in self.store:
if self.is_required(key):
missing.append(key)
else:
data[value["field"]] = self.store[key]
value = self.store[key]
if entry["flags"] & ResourceForm.LIST:
if type(value) is not list and type(value) is not tuple:
raise ValueError("Expected iterable value for key %s" % key)
data[field] = []
for v in value:
data[field].append(final_value(key, v))
else:
data[field] = final_value(key, value)
if missing:
raise RequirementError(missing)
......@@ -332,8 +198,15 @@ class ResourceForm(MutableMapping):
size = {
"Size": data["resourceTypeOption"]
}
elif data["type"]["displayName"] in ("rds", "rds-s3"):
raise RequirementError("rds or rds-s3 require a size parameter!")
data["resourceTypeOption"] = size
data["Visibility"] = {
"displayName": "Project Members",
"id": "8ab9c883-eb0d-4402-aaad-2e4007badce6"
}
return data
###############################################################################
\ No newline at end of file
from .coscine import CoscineClient
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
from .client import Client
from .project import Project
from .resource import Resource
from .object import Object
from .exceptions import *
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
###############################################################################
# This file contains module metadata and allows for a single point of truth
# regarding authorship and package version.
# It is used by setup.py and some files in 'src/'.
###############################################################################
__author__ = "RWTH Aachen University"
__version__ = "0.5.0"
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
###############################################################################
# This file contains a command line printable banner. It is printed to stdout
# on initialization of a client with verbose mode enabled.
# The banner includes the current package version and author and otherwise
# serves no special purpose.
###############################################################################
from .about import __version__, __author__
import colorama
###############################################################################
BANNER = """%s%s
_
(_)
___ ___ ___ ___ _ _ __ ___
/ __/ _ \/ __|/ __| | '_ \ / _ \
| (_| (_) \__ \ (__| | | | | __/
\___\___/|___/\___|_|_| |_|\___| %s
____________________________________
Coscine Python Client %s%s%s
Copyright (c) 2019-2021
%s
____________________________________%s
""" % (
colorama.Back.BLACK,
colorama.Fore.BLUE,
colorama.Fore.WHITE,
colorama.Fore.YELLOW,
__version__,
colorama.Fore.WHITE,
__author__,
colorama.Back.RESET
)
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
###############################################################################
# This file contains the main class of the coscine python package.
# The client class is the backbone and manager of everything contained
# within the coscine python module.
###############################################################################
import json
import urllib
import requests
from requests.exceptions import RequestException
import colorama
from requests.models import Response
from .banner import BANNER
from .about import __version__
from .exceptions import *
from .project import Project
from .static import StaticServer
from .ProjectForm import ProjectForm
###############################################################################
class Client:
#######################################
# Class member variable declaration
#######################################
lang: str
verbose: bool
session: requests.Session
loglevel: str
version: str
static: StaticServer
#######################################
# Class Constructor
###############################################################################
def __init__(self, token: str, lang: str = "en", verbose: bool = True, \
loglevel: list[str] = ["LOG", "INFO", "WARN", "REQUEST", "DATA"]):
LANG = ("en", "de")
if type(token) is not str:
raise TypeError("Argument 'token' must be of type string!")
if lang not in LANG:
raise ValueError("Invalid value in argument 'lang'!")
self.session = requests.Session()
self.session.headers = {
"Authorization": "Bearer " + token,
"User-Agent": "Coscine Python Client %s" % __version__
}
self.lang = lang
self.verbose = verbose
self.loglevel = loglevel
self.version = __version__
self.static = StaticServer(self)
if verbose:
colorama.init(autoreset=True)
print(BANNER)
###############################################################################
def log(self, msg: str, level: str = "LOG"):
LEVELS = {
"LOG": colorama.Fore.MAGENTA,
"INFO": colorama.Fore.GREEN,
"WARN": colorama.Fore.YELLOW,
"REQUEST": colorama.Fore.LIGHTGREEN_EX,
"DATA": colorama.Fore.LIGHTBLUE_EX
}
if level not in LEVELS:
raise ValueError(level)
if not level in self.loglevel:
return
if self.verbose:
print(LEVELS[level] + "[%s] " % level + msg)
###############################################################################
def is_outdated(self) -> bool:
uri = "https://pypi.org/pypi/coscine/json"
data = requests.get(uri).json()
version = data["info"]["version"]
if version != __version__:
msg = "Using Coscine Client version %s but latest version is %s.\n"\
"Consider updating the package by running:\n" \
"py -m pip install --user --upgrade coscine\n" \
% (__version__, version)
self.log(msg)
return True
else:
return False
###############################################################################
@staticmethod
def uri(api: str, endpoint: str, *args) -> str:
BASE = "https://coscine.rwth-aachen.de/coscine/api/Coscine.Api.%s/%s"
ENDPOINTS = (
"Blob",
"Metadata",
"Organization",
"Project",
"Resources",
"Tree",
"User",
"ActivatedFeatures"
)
if api not in ENDPOINTS:
raise ValueError("Argument 'api' does not specify a valid endpoint!")
uri = BASE % (api, endpoint)
for arg in args:
if arg is None:
continue
uri += "/" + urllib.parse.quote(arg, safe="")
return uri
###############################################################################
def _request(self, method: str, uri: str, **kwargs) -> Response:
if "data" in kwargs and type(kwargs["data"]) is dict:
kwargs["headers"] = {"Content-Type": "application/json;charset=utf-8"}
kwargs["data"] = json.dumps(kwargs["data"])
try:
self.log("%s %s" % (method, uri), "REQUEST")
if "data" in kwargs:
try:
self.log(json.dumps(json.loads(kwargs["data"]), indent=4), "DATA")
except :
pass
response = self.session.request(method, uri, **kwargs)
response.raise_for_status()
if response.content:
try:
self.log(json.dumps(response.json(), indent=4), "DATA")
except:
pass
return response
except requests.exceptions.ConnectionError:
raise ConnectionError()
except requests.exceptions.RequestException as e:
if e.response.status_code == 401:
raise UnauthorizedError("Invalid API token")
else:
raise RequestException()
###############################################################################
def get(self, uri: str, **kwargs) -> requests.Response:
return self._request("GET", uri, **kwargs)
###############################################################################
def put(self, uri: str, **kwargs) -> requests.Response:
return self._request("PUT", uri, **kwargs)
###############################################################################
def post(self, uri: str, **kwargs) -> requests.Response:
return self._request("POST", uri, **kwargs)
###############################################################################
def delete(self, uri: str, **kwargs) -> requests.Response:
return self._request("DELETE", uri, **kwargs)
###############################################################################
def projects(self, toplevel: bool = True, **kwargs) -> list[Project]:
ENDPOINTS = ("Project", "Project/-/topLevel")
uri = self.uri("Project", ENDPOINTS[toplevel])
projects = self.get(uri).json()
filter = []
for data in projects:
match = True
for key, value in kwargs.items():
if data[key] != value:
match = False
break
if match:
filter.append(Project(self, data))
return filter
###############################################################################
def project(self, displayName: str = None, **kwargs) -> Project:
if displayName:
kwargs["displayName"] = displayName
projects = self.projects(toplevel = True, **kwargs)
if len(projects) == 1:
return projects[0]
elif len(projects) == 0:
return None
else:
raise CoscineException("Undistinguishable projects!")
###############################################################################
def ProjectForm(self, parent: Project = None) -> ProjectForm:
vocabulary = {
"Discipline": self.static.disciplines(),
"Organization": self.static.organizations(),
"Visibility": self.static.visibility(),
"Features": self.static.features()
}
form = ProjectForm(self.lang, vocabulary, parent)
return form
###############################################################################
def create_project(self, form: ProjectForm) -> Project:
if type(form) is ProjectForm:
form = form.generate()
uri = self.uri("Project", "Project")
return Project(self, self.post(uri, data = form).json())
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python3 Client
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/wiki/-/tree/master/
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# This python wrapper implements a client for the Coscine API.
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.rwth-aachen.de for more information.
###############################################################################
import colorama
# Visit https://coscine.de for more information.
###############################################################################
class CoscineException(Exception):
def __init__(self, msg=None):
text = colorama.Fore.RED + "ERROR" + colorama.Fore.YELLOW
if msg:
text += ": " + msg
Exception.__init__(self, text)
pass
###############################################################################
class KeyError(CoscineException):
def __init__(self, msg=None):
CoscineException.__init__(self, msg)
###############################################################################
class ValueError(CoscineException):
def __init__(self, msg=None):
CoscineException.__init__(self, msg)
###############################################################################
class RequirementError(CoscineException):
def __init__(self, requirements):
msg = "Required field missing:\n%s" % "\n".join(requirements)
CoscineException.__init__(self, msg)
class ConnectionError(CoscineException):
pass
###############################################################################
class ServerError(CoscineException):
pass
class VocabularyError(CoscineException):
def __init__(self, value):
msg = "Value [%s] not in Vocabulary!" % value
CoscineException.__init__(self, msg)
###############################################################################
class NotFound(CoscineException):
def __init__(self, msg=None):
text = msg + " not found!"
CoscineException.__init__(self, text)
###############################################################################
class ConnectionError(CoscineException):
def __init__(self):
CoscineException.__init__(self, "Unable to reach Coscine!")
pass
###############################################################################
class RequirementError(CoscineException):
pass
class ServerError(CoscineException):
def __init__(self):
CoscineException.__init__(self)
\ No newline at end of file
class UnauthorizedError(CoscineException):
pass
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
from collections.abc import MutableMapping
from .exceptions import *
###############################################################################
class Form(MutableMapping):
# Form flags
NONE = 0
REQUIRED = 1
CONTROLLED = 2
SET = 4
FIXED = 8
LIST = 16
SPECIAL = 32
###############################################################################
def __init__(self, name: str, lang: str, entries: list, vocabulary: dict):
self._name = name
self._lang = lang
self._entries = entries
self._keys = {}
for entry in entries:
self._keys[entry["name"][self._lang]] = entry
self._vocabulary = vocabulary
self.store = {}
###############################################################################
def __getitem__(self, key: str) -> dict:
if key not in self.keys():
raise KeyError(key)
elif key in self.store:
return self.store[key]
else:
return None
###############################################################################
def __setitem__(self, key: str, value: object):
if key not in self._keys:
raise KeyError(key)
elif self.is_controlled(key):
vocabulary = self.get_vocabulary(key)
if type(value) is list:
self.store[key] = []
for val in value:
if val in vocabulary:
self.store[key].append(val)
else:
raise VocabularyError(val)
else:
if value in vocabulary:
self.store[key] = value
else:
raise VocabularyError(value)
else:
self.store[key] = value
###############################################################################
def __delitem__(self, key: str):
del self.store[key]
###############################################################################
def __iter__(self):
return iter(self.store)
###############################################################################
def __len__(self):
return len(self.store)
###############################################################################
def __str__(self) -> str:
entries = []
for key in self.keys():
R = " "
C = " "
value = ""
if self.is_required(key):
R = "R"
if self.is_controlled(key):
C = "C"
if key in self.store:
value = " = %s" % self.store[key]
entry = " [%s%s] %s%s" % (R, C, key, value)
entries.append(entry)
format = \
"_______________________________\n\n" \
" %s\n" \
"_______________________________\n\n" \
" [R: Required] [C: Controlled]\n" \
"-------------------------------\n" \
"%s\n" \
"_______________________________\n" \
% (self._name, "\n".join(entries))
return format
###############################################################################
def is_required(self, key: str) -> bool:
return self._keys[key]["flags"] & self.REQUIRED
###############################################################################
def is_controlled(self, key: str) -> bool:
return self._keys[key]["flags"] & self.CONTROLLED
###############################################################################
def get_field(self, key: str) -> str:
return self._keys[key]["field"]
###############################################################################
def get_vocabulary(self, key: str) -> dict:
if self.is_controlled(key):
return self._vocabulary[self.get_field(key)]
else:
msg = "Key [%s] is not controlled by a vocabulary!" % key
raise CoscineException(msg)
###############################################################################
def keys(self) -> list[str]:
return self._keys.keys()
###############################################################################
def reset(self):
"""
Resets the input form to default
"""
self.store.clear()
###############################################################################
def parse(self, data: dict):
pass
###############################################################################
def generate(self):
pass
###############################################################################
def entry(self, field: str) -> dict:
for item in self._entries:
if item["field"] == field:
return item
return None
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .client import Client
from .project import Project
###############################################################################
class Member:
# Member variables (only for type hinting, not static!)
###############################################################################
id: str
role: str
data: dict
client: Client
project: Project
###############################################################################
def __init__(self, project: Project, data: dict):
self.client = project.client
self.project = project
self.data = data
self.name = data["user"]["displayName"]
self.email = data["user"]["emailAddress"]
self.id = data["user"]["id"]
self.role = data["role"]["displayName"]
###############################################################################
def set_role(self, role: str):
ROLES = {
"Owner": "be294c5e-4e42-49b3-bec4-4b15f49df9a5",
"Member": "508b6d4e-c6ac-4aa5-8a8d-caa31dd39527"
}
if role not in ROLES:
raise ValueError("Invalid role '%s'." % role)
uri = self.client.uri("Project", "ProjectRole")
self.data["role"]["id"] = ROLES[role]
self.data["role"]["displayName"] = role
self.client.post(uri, data = self.data)
###############################################################################
def remove(self):
uri = self.client.uri("Project", "ProjectRole", "project", \
self.project.id, "user", self.id, "role", self.data["role"]["id"])
self.client.delete(uri)
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .client import Client
from .resource import Resource
from .MetadataForm import MetadataForm
import os
import json
from .progress import ProgressBar
###############################################################################
class Object:
###############################################################################
client: Client
resource: Resource
data: dict
metadata: dict
name: str
size: int
type: str
###############################################################################
CHUNK_SIZE = 4096
###############################################################################
def __init__(self, resource: Resource, data: dict, metadata: dict):
self.client = resource.client
self.resource = resource
self.data = data
self.metadata = metadata
self.name = data["Name"]
self.size = data["Size"]
self.type = data["Kind"]
###############################################################################
def __str__(self) -> str:
return json.dumps(self.data, indent = 4)
###############################################################################
def content(self) -> bytes:
uri = self.client.uri("Blob", "Blob", self.resource.id, self.name)
data = self.client.get(uri).content
return data
###############################################################################
def download(self, path = "./", callback = None):
# uri = self.data["Action"]["Download"] # Not working, Duh!
uri = self.client.uri("Blob", "Blob", self.resource.id, self.name)
response = self.client.get(uri, stream = True)
path = os.path.join(path, self.name)
fd = open(path, "wb")
bar = ProgressBar(self.client, self.size, self.name, "DOWN", callback)
for chunk in response.iter_content(chunk_size = self.CHUNK_SIZE):
fd.write(chunk)
bar.update(len(chunk))
fd.close()
###############################################################################
def delete(self):
uri = self.client.uri("Blob", "Blob", self.resource.id, self.name)
self.client.delete(uri)
###############################################################################
def update(self, metadata: MetadataForm):
if type(metadata) is MetadataForm:
metadata = metadata.generate()
uri = self.client.uri("Tree", "Tree", self.resource.id, self.name)
self.client.put(uri, data = metadata)
###############################################################################
def MetadataForm(self) -> MetadataForm:
form = self.resource.MetadataForm()
form.parse(self.metadata)
return form
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
###############################################################################
# This file provides the ProgressBar class - a simple wrapper around tqdm
# progress bars. It is used in download/upload methods and provides the
# benefit of remembering state information and printing only when in
# verbose mode.
###############################################################################
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from coscine.client import Client
from tqdm import tqdm
###############################################################################
class ProgressBar:
def __init__(self, client: Client, filesize: int, filename: str, \
mode: str, callafter: function = None):
MODES = {
"UP": "",
"DOWN": ""
}
if mode not in MODES:
raise ValueError(mode)
self.client = client
if client.verbose:
self.bar = tqdm(total = filesize, unit = "B", unit_scale = True,
desc = "%s %s" % (MODES[mode], filename))
self.callafter = callafter
self.n = 0
###############################################################################
def update(self, chunksize: int):
self.n += chunksize
if self.client.verbose:
self.bar.update(chunksize)
if self.callafter:
self.callafter(chunksize)
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .client import Client
import os
import json
from .exceptions import *
from .resource import Resource
from .ResourceForm import ResourceForm
from .ProjectForm import ProjectForm
from .MetadataPresetForm import MetadataPresetForm
from .member import Member
###############################################################################
class Project:
#######################################
# Class member variable declaration
#######################################
client: Client
data: dict
id: str
name: str
description: str
principleInvestigators: str
startDate: str
endDate: str
discipline: list[str]
organization: list[str]
#######################################
def __init__(self, client: Client, data: dict):
self.client = client
self.data = data
self.id = data["id"]
self.name = data["displayName"]
self.description = data["description"]
self.principleInvestigators = data["principleInvestigators"]
self.startDate = data["startDate"]
self.endDate = data["endDate"]
self.discipline = []
self.organization = []
###############################################################################
def __str__(self) -> str:
return json.dumps(self.data, indent=4)
###############################################################################
def subprojects(self, **kwargs) -> list:
uri = self.client.uri("Project", "SubProject", self.id)
projects = self.client.get(uri).json()
filter = []
for data in projects:
match = True
for key, value in kwargs.items():
if data[key] != value:
match = False
break
if match:
filter.append(Project(self.client, data))
return filter
###############################################################################
def resources(self, **kwargs) -> list[Resource]:
uri = self.client.uri("Project", "Project", self.id, "resources")
resources = self.client.get(uri).json()
filter = []
for data in resources:
match = True
for key, value in kwargs.items():
if data[key] != value:
match = False
break
if match:
filter.append(Resource(self, data))
return filter
###############################################################################
def resource(self, displayName: str = None, **kwargs) -> Resource:
if displayName:
kwargs["displayName"] = displayName
resources = self.resources(**kwargs)
if len(resources) == 1:
return resources[0]
else:
raise CoscineException("Undistinguishable resources!")
###############################################################################
def download(self, path: str = "./"):
path = os.path.join(path, self.displayName)
if not os.path.isdir(path):
os.mkdir(path)
for resource in self.resources():
resource.download(path = path)
###############################################################################
def delete(self):
uri = self.client.uri("Project", "Project", self.id)
self.client.delete(uri)
###############################################################################
def members(self) -> list[Member]:
uri = self.client.uri("Project", "ProjectRole", self.id)
data = self.client.get(uri).json()
members = [Member(self, m) for m in data]
return members
###############################################################################
def invite(self, email: str, role: str = "Member"):
ROLES = {
"Owner": "be294c5e-4e42-49b3-bec4-4b15f49df9a5",
"Member": "508b6d4e-c6ac-4aa5-8a8d-caa31dd39527"
}
if role not in ROLES:
raise ValueError("Invalid role '%s'." % role)
uri = self.client.uri("Project", "Project", "invitation")
data = {
"projectId": self.data["id"],
"role": ROLES[role],
"email": email
}
try:
self.client.log("Inviting [%s] as [%s] to project [%s]." %
(email, role, self.id))
self.client.post(uri, data = data)
except ServerError:
self.client.log("User [%s] has invite pending." % email)
###############################################################################
def _instance(self, link: str) -> dict:
uri = self.client.uri("Metadata", "Metadata", "instances", self.id, link)
return self.client.get(uri).json()
###############################################################################
def ResourceForm(self) -> ResourceForm:
vocabulary = {
"type": self.client.static.resource_types(),
"applicationProfile": self.client.static.application_profiles(),
"License": self.client.static.licenses(),
"Visibility": self.client.static.visibility(),
"Disciplines": self.client.static.disciplines()
}
return ResourceForm(self.client.lang, vocabulary)
###############################################################################
def create_resource(self, resourceForm: ResourceForm, \
metadataPreset: MetadataPresetForm = None) -> Resource:
if type(resourceForm) is ResourceForm:
resourceForm = resourceForm.generate()
if metadataPreset:
if type(metadataPreset) is MetadataPresetForm:
metadataPreset = metadataPreset.generate()
resourceForm["fixedValues"] = metadataPreset
uri = self.client.uri("Resources", "Resource", "Project", self.id)
return Resource(self, self.client.post(uri, data = resourceForm).json())
###############################################################################
def form(self) -> ProjectForm:
form = self.client.ProjectForm()
form.parse(self.data)
return form
###############################################################################
def update(self, form: ProjectForm):
if type(form) is ProjectForm:
form = form.generate()
uri = self.client.uri("Project", "Project", self.id)
self.client.post(uri, data = form)
###############################################################################
\ No newline at end of file
###############################################################################
# Coscine Python Client
# Copyright (c) 2018-2021 RWTH Aachen University
# Contact: coscine@itc.rwth-aachen.de
# Git: https://git.rwth-aachen.de/coscine/docs/public/coscine-python-client
# Please direct bug reports, feature requests or questions at the URL above
# by opening an issue.
###############################################################################
# Coscine is an open source project at RWTH Aachen University for
# the management of research data.
# Visit https://coscine.de for more information.
###############################################################################
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .client import Client
from .project import Project
import os
import json
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
from .exceptions import *
from .object import Object
from .MetadataForm import MetadataForm
from .MetadataPresetForm import MetadataPresetForm
from .ResourceForm import ResourceForm
from .progress import ProgressBar
###############################################################################
class Resource:
client: Client
project: Project
data: dict
id: str
name: str
def __init__(self, project, data):
self.client = project.client
self.project = project
self.data = data
self.id = data["id"]
self.name = data["displayName"]
###############################################################################
def __str__(self) -> str:
return json.dumps(self.data, indent=4)
###############################################################################
def upload(self, key: str, file, metadata: dict = None, callback: function = None):
if hasattr(file, "read"):
fd = file
filename = "MEM"
elif type(file) is str:
fd = open(file, "rb")
filename = file
else:
raise CoscineException()
if metadata:
if type(metadata) is MetadataForm:
metadata = metadata.generate()
uri = self.client.uri("Tree", "Tree", self.id, key)
self.client.put(uri, data = metadata)
uri = self.client.uri("Blob", "Blob", self.id, key)
fields = {"files": (key, fd, "application/octect-stream")}
encoder = MultipartEncoder(fields = fields)
bar = ProgressBar(self.client, encoder.len, filename, "UP", callback)
monitor = MultipartEncoderMonitor(encoder, callback = lambda monitor: \
bar.update(monitor.bytes_read - bar.n))
headers = {"Content-Type": monitor.content_type}
self.client.put(uri, data = monitor, headers = headers)
###############################################################################
def download(self, path = "./"):
path = os.path.join(path, self.name)
if not os.path.isdir(path):
os.mkdir(path)
for file in self.objects():
file.download(path)
###############################################################################
def delete(self):
uri = self.client.uri("Resources", "Resource", self.id)
self.client.delete(uri)
###############################################################################
def quota(self) -> int:
uri = self.client.uri("Blob", "Blob", self.id, "quota")
data = self.client.get(uri).json()
return int(data("data")["usedSizeByte"])
###############################################################################
def objects(self, **kwargs) -> list[Object]:
uri = self.client.uri("Tree", "Tree", self.id)
data = self.client.get(uri).json()
fileStorage = data["data"]["fileStorage"]
metadataStorage = data["data"]["metadataStorage"]
objects = zip(fileStorage, metadataStorage)
filter = []
for f in objects:
match = True
for key, value in kwargs.items():
if f[0][key] != value:
match = False
break
if match:
filter.append(Object(self, f[0], f[1]))
return filter
###############################################################################
def object(self, displayName: str = None) -> Object:
objects = self.objects()
for it in objects:
if it.name == displayName:
return it
return None
###############################################################################
def applicationProfile(self, parse = False):
return self.client.static.application_profile(
self.data["applicationProfile"], self.id, parse)
###############################################################################
def MetadataForm(self, data = None):
entries = []
vocabulary = {}
lang = self.client.lang
profile = self.applicationProfile(parse = True)
for element in profile["graph"]:
flags = MetadataForm.NONE
if "minCount" in element and element["minCount"] > 0:
flags |= MetadataForm.REQUIRED
if "maxCount" in element and element["maxCount"] > 1:
flags |= MetadataForm.LIST
if "class" in element:
flags |= MetadataForm.CONTROLLED
uri = element["class"]
instance = self.project._instance(uri)
if lang not in instance or len(instance[lang]) == 0:
lang = "en"
voc = {}
for rule in instance[lang]:
voc[rule["name"]] = rule["value"]
vocabulary[element["name"][lang]] = voc
entry = {
"name": element["name"],
"path": element["path"],
"field": element["name"][lang],
"flags": flags,
"order": element["order"],
"datatype": element["datatype"],
"type": element["type"]
}
entries.append(entry)
# Sort the keys according to their application profile order
entries = sorted(entries, key = lambda x: x["order"])
form = MetadataForm(profile, self.client.lang, entries, vocabulary)
if data:
for key in data:
form[key] = data[key]
return form
###############################################################################
def MetadataPresetForm(self) -> MetadataPresetForm:
entries = []
vocabulary = {}
lang = self.client.lang
applicationProfile = self.applicationProfile(parse = True)
for element in applicationProfile["graph"]:
flags = MetadataPresetForm.NONE
if "minCount" in element and element["minCount"] > 0:
flags |= MetadataPresetForm.REQUIRED
if "maxCount" in element and element["maxCount"] > 1:
flags |= MetadataPresetForm.LIST
if "class" in element:
flags |= MetadataPresetForm.CONTROLLED
uri = element["class"]
instance = self.project._instance(uri)
if lang not in instance or len(instance[lang]) == 0:
lang = "en"
voc = {}
for rule in instance[lang]:
voc[rule["name"]] = rule["value"]
vocabulary[element["name"][lang]] = voc
entry = {
"name": element["name"],
"path": element["id"],
"flags": flags,
"field": element["name"][lang],
"order": element["order"],
"datatype": element["datatype"],
"type": element["type"],
"locked": False,
"enabled": True
}
entries.append(entry)
# Sort the keys according to their application profile order
entries = sorted(entries, key = lambda x: x["order"])
return MetadataPresetForm(applicationProfile, lang, entries, vocabulary)
###############################################################################
def form(self) -> ResourceForm:
form = self.project.ResourceForm()
form.parse(self.data)
return form
###############################################################################
def update(self, form):
if type(form) is ResourceForm:
form = form.generate()
uri = self.client.uri("Resources", "Resource", self.id)
self.client.post(uri, data = form)
###############################################################################
def update_preset(self, form):
data = self.form().generate()
data["fixedValues"] = form.generate()
uri = self.client.uri("Resources", "Resource", self.id)
self.client.post(uri, data = data)
###############################################################################
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment