Module coscine.resource

This file defines the resource object for the representation of Coscine resources. It provides an easy interface to interact with Coscine resources from python.

Expand source code
###############################################################################
# Coscine Python SDK
# Copyright (c) 2018-2022 RWTH Aachen University
# Licensed under the terms of the MIT License
# #############################################################################
# Coscine, short for Collaborative Scientific Integration Environment is
# a platform for research data management (RDM).
# For more information on Coscine visit https://www.coscine.de/.
#
# Please note that this python module is open source software primarily
# developed and maintained by the scientific community. It is not
# an official service that RWTH Aachen provides support for.
###############################################################################

###############################################################################
# File description
###############################################################################

"""
This file defines the resource object for the representation of
Coscine resources. It provides an easy interface to interact with Coscine
resources from python.
"""

###############################################################################
# Dependencies
###############################################################################

from __future__ import annotations
from typing import TYPE_CHECKING, List, Callable
if TYPE_CHECKING:
        from .client import Client
        from .project import Project
from .exceptions import *
from .object import FileObject, MetadataForm
from .logger import ProgressBar
from .form import InputForm
from .graph import ApplicationProfile
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
from prettytable.prettytable import PrettyTable
from concurrent.futures import ThreadPoolExecutor, wait
import json
import os

###############################################################################
# Class definition
###############################################################################

class S3info:
        """
        Provides a simple pythonic interface to resource S3 access data
        """

        read_access_key: str
        read_secret_key: str
        write_access_key: str
        write_secret_key: str
        endpoint: str
        bucket: str
        _data: dict

        def __init__(self, data: dict) -> None:
                self._data = data
                if "resourceTypeOption" in data and data["resourceTypeOption"]:
                        opt: dict = data["resourceTypeOption"]
                        self.read_access_key = opt.get("ReadAccessKey")
                        self.read_secret_key = opt.get("ReadSecretKey")
                        self.write_access_key = opt.get("WriteAccessKey")
                        self.write_secret_key = opt.get("WriteSecretKey")
                        self.endpoint = opt.get("Endpoint")
                        self.bucket = opt.get("BucketName")

###############################################################################
# Class definition
###############################################################################

class ResourceForm(InputForm):
        """
        """

###############################################################################

        def __init__(self, client: Client) -> None:
                self._items = client.vocabularies.builtin("resource")
                super().__init__(client)
                vocabularies = {
                        "type": client.vocabularies.resource_types(True),
                        "applicationProfile": client.vocabularies.application_profiles(True),
                        "license": client.vocabularies.licenses(True),
                        "visibility": client.vocabularies.visibility(True),
                        "disciplines": client.vocabularies.disciplines(True)
                }
                for item in self._items:
                        if item.vocabulary:
                                self._vocabularies[item.path] = vocabularies[item.path]

###############################################################################

        def parse(self, data: dict) -> None:
                IGNORE = ["id", "pid", "fixedValues", "creator", "archived"]
                if data is None:
                        return
                for path, value in data.items():
                        if path in IGNORE: continue
                        key = self.name_of(path)
                        self.set_value(key, value, True)

###############################################################################

        def generate(self) -> dict:
                metadata = {}

                # Collect missing required fields
                missing = []

                # Set metadata
                for key, values in self.items(True):
                        
                        if not values:
                                if self.is_required(key):
                                        missing.append(key)
                                continue

                        properties = self.properties(key)
                        metadata[properties.path] = values if len(values) > 1 else values[0]

                # Check for missing required fields
                if len(missing) > 0:
                        if len(missing) == 1 and missing[0] == "resourceTypeOption" \
                                                                        and     metadata["type"] == "linked":
                                pass
                        else:
                                raise ValueError(missing)

                return metadata

###############################################################################
# Class definition
###############################################################################

class Resource:
        """
        Python representation of a Coscine Resource type.
        """

        client: Client
        project: Project
        data: dict

        id: str
        pid: str
        name: str
        displayName: str
        type: str
        disciplines: List[str]
        profile: str
        archived: bool
        creator: str
        s3: S3info

###############################################################################

        def __init__(self, project: Project, data: dict) -> None:
                """
                Initializes a Coscine resource object.

                Parameters
                ----------
                project : Project
                        Coscine project handle
                data : dict
                        Resource data received from Coscine.
                """

                self.project = project
                self.client = self.project.client
                self.data = data
                self.s3 = S3info(data)

###############################################################################

        @property
        def id(self) -> str: return self.data["id"]

        @property
        def pid(self) -> str: return self.data["pid"]

        @property
        def name(self) -> str: return self.data["resourceName"]

        @property
        def displayName(self) -> str: return self.data["displayName"]

        @property
        def license(self) -> str:
                return self.data["license"]["displayName"] \
                        if self.data["license"] else ""

        @property
        def type(self) -> str: return self.data["type"]["displayName"]

        @property
        def disciplines(self) -> List[str]:
                lang = {
                        "en": "displayNameEn",
                        "de": "displayNameDe"
                }[self.client.language]
                return [k[lang] for k in self.data["disciplines"]]

        @property
        def profile(self) -> str: return self.data["applicationProfile"]

        @property
        def archived(self) -> bool: return bool(self.data["archived"])

        @property
        def creator(self) -> str: return self.data["creator"]

###############################################################################

        def __str__(self) -> str:
                table = PrettyTable(["Property", "Value"])
                rows = [
                        ("ID", self.id),
                        ("Resource Name", self.name),
                        ("Display Name", self.displayName),
                        ("PID", self.pid),
                        ("Type", self.type),
                        ("Disciplines", "\n".join(self.disciplines)),
                        ("License", self.license),
                        ("Application Profile", self.profile),
                        ("Archived", self.archived),
                        ("Creator", self.creator),
                        ("Project", self.project.displayName),
                        ("Project ID", self.project.id)
                ]
                table.max_width["Value"] = 50
                table.add_rows(rows)
                return table.get_string(title = "Resource [%s]" % self.displayName)

###############################################################################

        def delete(self) -> None:
                """
                Deletes the Coscine resource and all objects contained within it on
                the Coscine servers.
                """

                uri = self.client.uri("Resources", "Resource", self.id)
                self.client.delete(uri)

###############################################################################

        def application_profile(self) -> ApplicationProfile:
                """
                Returns the application profile of the resource
                """

                return self.client.vocabularies.application_profile(self.profile)

###############################################################################

        def _download_concurrently(self, path: str = "./") -> None:
                with ThreadPoolExecutor(max_workers=4) as executor:
                        files = self.objects()
                        futures = [executor.submit(file.download, path) for file in files]
                        wait(futures)
                        for index, fut in enumerate(futures):
                                try: fut.result()
                                except CoscineException:
                                        self.client.logger.warn(
                                                f"Error downloading '{files[index].name}'.")

###############################################################################

        def download(self, path: str = "./", metadata: bool = False) -> None:
                """
                Downloads the resource and all of its contents to the local harddrive.

                Parameters
                ----------
                path : str, default: "./"
                        Path to the local storage location.
                metadata : bool, default: False
                        If enabled, resource metadata is downloaded and put in
                        a hidden file '.metadata.json'.
                """

                path = os.path.join(path, self.name)
                if not os.path.isdir(path):
                        os.mkdir(path)
                if self.client.concurrent:
                        self._download_concurrently(path)
                else:
                        for file in self.objects():
                                try: file.download(path)
                                except CoscineException:
                                        self.client.logger.warn(f"Error downloading '{file.name}'.")
                if metadata:
                        data = json.dumps(self.data, indent=4)
                        with open(os.path.join(path, ".resource-metadata.json"), "w") as fd:
                                fd.write(data)

###############################################################################

        def objects(self, path: str = None, **kwargs) -> List[FileObject]:
                """
                Returns a list of Objects stored within the resource

                Parameters
                ------------
                kwargs
                        file-object filter arguments (e.g. 'Name' = 'testfile').

                Returns
                -------
                list[Object]
                        List of Coscine file-like objects.
                """

                objects = []
                uri = self.client.uri("Tree", "Tree", self.id)
                if path:
                        args = {"path": path}
                else:
                        args = None
                data = self.client.get(uri, params = args).json()
                fileStorage = data["data"]["fileStorage"]
                metadataStorage = data["data"]["metadataStorage"]
                for data in fileStorage:
                        for key, value in kwargs.items():
                                if data[key] != value:
                                        break
                        else:
                                objects.append(FileObject(self, data))
                return objects

###############################################################################

        def object(self, path: str, **kwargs) -> FileObject:
                """
                Returns an Object stored within the resource

                Parameters
                ------------
                displayName : str, default: None
                        file-object display name (filename/key).
                kwargs
                        Filter

                Returns
                -------
                FileObject
                        Python representation of the file-object as an Object instance
                """
                
                objects = self.objects(Path=path, **kwargs)
                if len(objects) == 1:
                        return objects[0]
                elif len(objects) == 0:
                        return None
                else:
                        raise ValueError("Found more than 1 FileObject matching "\
                                                                                        "the specified criteria!")

###############################################################################

        def upload(self, key: str, file, metadata = None, \
                                        callback: Callable[[int]] = None) -> None:
                """
                Uploads a file-like object to a resource on the Coscine server

                Parameters
                ----------
                key : str
                        filename of the file-like object.
                file : object with read() attribute
                                Either open file handle or local file location path.
                metadata : dict
                        File metadata. For rds-s3 this is optional, but recommended.
                callback : Callable[int]
                        Optional callback called during chunk uploads.
                """

                if hasattr(file, "read"):
                        fd = file
                        filename = "MEM"
                elif type(file) is str:
                        fd = open(file, "rb")
                        filename = file
                else:
                        raise TypeError("Argument `file` has unexpected type!")

                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.logger, 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 set_archived(self, flag: bool) -> None:
                """
                Set the archived flag of the resource to put it in read-only mode.
                Only the resource creator or project owner can do this.

                Parameters
                ----------
                flag : bool
                        Enable with True, Disable with False.
                """

                uri = self.client.uri("Resources", "Resource", self.id, \
                                        "setReadonly?status=%s" % str(flag).lower())
                self.client.post(uri)
                self.data["archived"] = flag
                self.archived = flag

###############################################################################

        def form(self) -> ResourceForm:
                """
                Returns a ResourceForm filled with the metadata of the current resource.

                Returns
                -------
                ResourceForm
                """

                form = self.client.forms.resource()
                form.parse(self.data)
                return form

###############################################################################

        def update(self, form: ResourceForm) -> dict:
                """
                Updates the metadata of the resource using the supplied ResourceForm.

                Parameters
                ----------
                form : ResourceForm
                        ResourceForm filled with updated values.
                """

                if type(form) is ResourceForm:
                        form = form.generate()
                elif type(form) is not dict:
                        raise TypeError("")

                uri = self.client.uri("Resources", "Resource", self.id)
                response = self.client.post(uri, data = form)
                if response.ok: self.data = response.json()
                return response

###############################################################################

        def metadata(self, data: dict = None) -> MetadataForm:
                """
                Creates a MetadataForm for this resource
                """

                return self.client.forms.metadata(self.application_profile())

###############################################################################

Classes

class Resource (project: Project, data: dict)

Python representation of a Coscine Resource type.

Initializes a Coscine resource object.

Parameters

project : Project
Coscine project handle
data : dict
Resource data received from Coscine.
Expand source code
class Resource:
        """
        Python representation of a Coscine Resource type.
        """

        client: Client
        project: Project
        data: dict

        id: str
        pid: str
        name: str
        displayName: str
        type: str
        disciplines: List[str]
        profile: str
        archived: bool
        creator: str
        s3: S3info

###############################################################################

        def __init__(self, project: Project, data: dict) -> None:
                """
                Initializes a Coscine resource object.

                Parameters
                ----------
                project : Project
                        Coscine project handle
                data : dict
                        Resource data received from Coscine.
                """

                self.project = project
                self.client = self.project.client
                self.data = data
                self.s3 = S3info(data)

###############################################################################

        @property
        def id(self) -> str: return self.data["id"]

        @property
        def pid(self) -> str: return self.data["pid"]

        @property
        def name(self) -> str: return self.data["resourceName"]

        @property
        def displayName(self) -> str: return self.data["displayName"]

        @property
        def license(self) -> str:
                return self.data["license"]["displayName"] \
                        if self.data["license"] else ""

        @property
        def type(self) -> str: return self.data["type"]["displayName"]

        @property
        def disciplines(self) -> List[str]:
                lang = {
                        "en": "displayNameEn",
                        "de": "displayNameDe"
                }[self.client.language]
                return [k[lang] for k in self.data["disciplines"]]

        @property
        def profile(self) -> str: return self.data["applicationProfile"]

        @property
        def archived(self) -> bool: return bool(self.data["archived"])

        @property
        def creator(self) -> str: return self.data["creator"]

###############################################################################

        def __str__(self) -> str:
                table = PrettyTable(["Property", "Value"])
                rows = [
                        ("ID", self.id),
                        ("Resource Name", self.name),
                        ("Display Name", self.displayName),
                        ("PID", self.pid),
                        ("Type", self.type),
                        ("Disciplines", "\n".join(self.disciplines)),
                        ("License", self.license),
                        ("Application Profile", self.profile),
                        ("Archived", self.archived),
                        ("Creator", self.creator),
                        ("Project", self.project.displayName),
                        ("Project ID", self.project.id)
                ]
                table.max_width["Value"] = 50
                table.add_rows(rows)
                return table.get_string(title = "Resource [%s]" % self.displayName)

###############################################################################

        def delete(self) -> None:
                """
                Deletes the Coscine resource and all objects contained within it on
                the Coscine servers.
                """

                uri = self.client.uri("Resources", "Resource", self.id)
                self.client.delete(uri)

###############################################################################

        def application_profile(self) -> ApplicationProfile:
                """
                Returns the application profile of the resource
                """

                return self.client.vocabularies.application_profile(self.profile)

###############################################################################

        def _download_concurrently(self, path: str = "./") -> None:
                with ThreadPoolExecutor(max_workers=4) as executor:
                        files = self.objects()
                        futures = [executor.submit(file.download, path) for file in files]
                        wait(futures)
                        for index, fut in enumerate(futures):
                                try: fut.result()
                                except CoscineException:
                                        self.client.logger.warn(
                                                f"Error downloading '{files[index].name}'.")

###############################################################################

        def download(self, path: str = "./", metadata: bool = False) -> None:
                """
                Downloads the resource and all of its contents to the local harddrive.

                Parameters
                ----------
                path : str, default: "./"
                        Path to the local storage location.
                metadata : bool, default: False
                        If enabled, resource metadata is downloaded and put in
                        a hidden file '.metadata.json'.
                """

                path = os.path.join(path, self.name)
                if not os.path.isdir(path):
                        os.mkdir(path)
                if self.client.concurrent:
                        self._download_concurrently(path)
                else:
                        for file in self.objects():
                                try: file.download(path)
                                except CoscineException:
                                        self.client.logger.warn(f"Error downloading '{file.name}'.")
                if metadata:
                        data = json.dumps(self.data, indent=4)
                        with open(os.path.join(path, ".resource-metadata.json"), "w") as fd:
                                fd.write(data)

###############################################################################

        def objects(self, path: str = None, **kwargs) -> List[FileObject]:
                """
                Returns a list of Objects stored within the resource

                Parameters
                ------------
                kwargs
                        file-object filter arguments (e.g. 'Name' = 'testfile').

                Returns
                -------
                list[Object]
                        List of Coscine file-like objects.
                """

                objects = []
                uri = self.client.uri("Tree", "Tree", self.id)
                if path:
                        args = {"path": path}
                else:
                        args = None
                data = self.client.get(uri, params = args).json()
                fileStorage = data["data"]["fileStorage"]
                metadataStorage = data["data"]["metadataStorage"]
                for data in fileStorage:
                        for key, value in kwargs.items():
                                if data[key] != value:
                                        break
                        else:
                                objects.append(FileObject(self, data))
                return objects

###############################################################################

        def object(self, path: str, **kwargs) -> FileObject:
                """
                Returns an Object stored within the resource

                Parameters
                ------------
                displayName : str, default: None
                        file-object display name (filename/key).
                kwargs
                        Filter

                Returns
                -------
                FileObject
                        Python representation of the file-object as an Object instance
                """
                
                objects = self.objects(Path=path, **kwargs)
                if len(objects) == 1:
                        return objects[0]
                elif len(objects) == 0:
                        return None
                else:
                        raise ValueError("Found more than 1 FileObject matching "\
                                                                                        "the specified criteria!")

###############################################################################

        def upload(self, key: str, file, metadata = None, \
                                        callback: Callable[[int]] = None) -> None:
                """
                Uploads a file-like object to a resource on the Coscine server

                Parameters
                ----------
                key : str
                        filename of the file-like object.
                file : object with read() attribute
                                Either open file handle or local file location path.
                metadata : dict
                        File metadata. For rds-s3 this is optional, but recommended.
                callback : Callable[int]
                        Optional callback called during chunk uploads.
                """

                if hasattr(file, "read"):
                        fd = file
                        filename = "MEM"
                elif type(file) is str:
                        fd = open(file, "rb")
                        filename = file
                else:
                        raise TypeError("Argument `file` has unexpected type!")

                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.logger, 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 set_archived(self, flag: bool) -> None:
                """
                Set the archived flag of the resource to put it in read-only mode.
                Only the resource creator or project owner can do this.

                Parameters
                ----------
                flag : bool
                        Enable with True, Disable with False.
                """

                uri = self.client.uri("Resources", "Resource", self.id, \
                                        "setReadonly?status=%s" % str(flag).lower())
                self.client.post(uri)
                self.data["archived"] = flag
                self.archived = flag

###############################################################################

        def form(self) -> ResourceForm:
                """
                Returns a ResourceForm filled with the metadata of the current resource.

                Returns
                -------
                ResourceForm
                """

                form = self.client.forms.resource()
                form.parse(self.data)
                return form

###############################################################################

        def update(self, form: ResourceForm) -> dict:
                """
                Updates the metadata of the resource using the supplied ResourceForm.

                Parameters
                ----------
                form : ResourceForm
                        ResourceForm filled with updated values.
                """

                if type(form) is ResourceForm:
                        form = form.generate()
                elif type(form) is not dict:
                        raise TypeError("")

                uri = self.client.uri("Resources", "Resource", self.id)
                response = self.client.post(uri, data = form)
                if response.ok: self.data = response.json()
                return response

###############################################################################

        def metadata(self, data: dict = None) -> MetadataForm:
                """
                Creates a MetadataForm for this resource
                """

                return self.client.forms.metadata(self.application_profile())

Class variables

var client : Client
var data : dict
var project : Project
var s3S3info

Instance variables

var archived : bool
Expand source code
@property
def archived(self) -> bool: return bool(self.data["archived"])
var creator : str
Expand source code
@property
def creator(self) -> str: return self.data["creator"]
var disciplines : List[str]
Expand source code
@property
def disciplines(self) -> List[str]:
        lang = {
                "en": "displayNameEn",
                "de": "displayNameDe"
        }[self.client.language]
        return [k[lang] for k in self.data["disciplines"]]
var displayName : str
Expand source code
@property
def displayName(self) -> str: return self.data["displayName"]
var id : str
Expand source code
@property
def id(self) -> str: return self.data["id"]
var license : str
Expand source code
@property
def license(self) -> str:
        return self.data["license"]["displayName"] \
                if self.data["license"] else ""
var name : str
Expand source code
@property
def name(self) -> str: return self.data["resourceName"]
var pid : str
Expand source code
@property
def pid(self) -> str: return self.data["pid"]
var profile : str
Expand source code
@property
def profile(self) -> str: return self.data["applicationProfile"]
var type : str
Expand source code
@property
def type(self) -> str: return self.data["type"]["displayName"]

Methods

def application_profile(self) ‑> ApplicationProfile

Returns the application profile of the resource

Expand source code
def application_profile(self) -> ApplicationProfile:
        """
        Returns the application profile of the resource
        """

        return self.client.vocabularies.application_profile(self.profile)
def delete(self) ‑> None

Deletes the Coscine resource and all objects contained within it on the Coscine servers.

Expand source code
def delete(self) -> None:
        """
        Deletes the Coscine resource and all objects contained within it on
        the Coscine servers.
        """

        uri = self.client.uri("Resources", "Resource", self.id)
        self.client.delete(uri)
def download(self, path: str = './', metadata: bool = False) ‑> None

Downloads the resource and all of its contents to the local harddrive.

Parameters

path : str, default: "./"
Path to the local storage location.
metadata : bool, default: False
If enabled, resource metadata is downloaded and put in a hidden file '.metadata.json'.
Expand source code
def download(self, path: str = "./", metadata: bool = False) -> None:
        """
        Downloads the resource and all of its contents to the local harddrive.

        Parameters
        ----------
        path : str, default: "./"
                Path to the local storage location.
        metadata : bool, default: False
                If enabled, resource metadata is downloaded and put in
                a hidden file '.metadata.json'.
        """

        path = os.path.join(path, self.name)
        if not os.path.isdir(path):
                os.mkdir(path)
        if self.client.concurrent:
                self._download_concurrently(path)
        else:
                for file in self.objects():
                        try: file.download(path)
                        except CoscineException:
                                self.client.logger.warn(f"Error downloading '{file.name}'.")
        if metadata:
                data = json.dumps(self.data, indent=4)
                with open(os.path.join(path, ".resource-metadata.json"), "w") as fd:
                        fd.write(data)
def form(self) ‑> ResourceForm

Returns a ResourceForm filled with the metadata of the current resource.

Returns

ResourceForm
 
Expand source code
def form(self) -> ResourceForm:
        """
        Returns a ResourceForm filled with the metadata of the current resource.

        Returns
        -------
        ResourceForm
        """

        form = self.client.forms.resource()
        form.parse(self.data)
        return form
def metadata(self, data: dict = None) ‑> MetadataForm

Creates a MetadataForm for this resource

Expand source code
def metadata(self, data: dict = None) -> MetadataForm:
        """
        Creates a MetadataForm for this resource
        """

        return self.client.forms.metadata(self.application_profile())
def object(self, path: str, **kwargs) ‑> FileObject

Returns an Object stored within the resource

Parameters

displayName : str, default: None
file-object display name (filename/key).
kwargs
Filter

Returns

FileObject
Python representation of the file-object as an Object instance
Expand source code
def object(self, path: str, **kwargs) -> FileObject:
        """
        Returns an Object stored within the resource

        Parameters
        ------------
        displayName : str, default: None
                file-object display name (filename/key).
        kwargs
                Filter

        Returns
        -------
        FileObject
                Python representation of the file-object as an Object instance
        """
        
        objects = self.objects(Path=path, **kwargs)
        if len(objects) == 1:
                return objects[0]
        elif len(objects) == 0:
                return None
        else:
                raise ValueError("Found more than 1 FileObject matching "\
                                                                                "the specified criteria!")
def objects(self, path: str = None, **kwargs) ‑> List[FileObject]

Returns a list of Objects stored within the resource

Parameters

kwargs
file-object filter arguments (e.g. 'Name' = 'testfile').

Returns

list[Object]
List of Coscine file-like objects.
Expand source code
def objects(self, path: str = None, **kwargs) -> List[FileObject]:
        """
        Returns a list of Objects stored within the resource

        Parameters
        ------------
        kwargs
                file-object filter arguments (e.g. 'Name' = 'testfile').

        Returns
        -------
        list[Object]
                List of Coscine file-like objects.
        """

        objects = []
        uri = self.client.uri("Tree", "Tree", self.id)
        if path:
                args = {"path": path}
        else:
                args = None
        data = self.client.get(uri, params = args).json()
        fileStorage = data["data"]["fileStorage"]
        metadataStorage = data["data"]["metadataStorage"]
        for data in fileStorage:
                for key, value in kwargs.items():
                        if data[key] != value:
                                break
                else:
                        objects.append(FileObject(self, data))
        return objects
def set_archived(self, flag: bool) ‑> None

Set the archived flag of the resource to put it in read-only mode. Only the resource creator or project owner can do this.

Parameters

flag : bool
Enable with True, Disable with False.
Expand source code
def set_archived(self, flag: bool) -> None:
        """
        Set the archived flag of the resource to put it in read-only mode.
        Only the resource creator or project owner can do this.

        Parameters
        ----------
        flag : bool
                Enable with True, Disable with False.
        """

        uri = self.client.uri("Resources", "Resource", self.id, \
                                "setReadonly?status=%s" % str(flag).lower())
        self.client.post(uri)
        self.data["archived"] = flag
        self.archived = flag
def update(self, form: ResourceForm) ‑> dict

Updates the metadata of the resource using the supplied ResourceForm.

Parameters

form : ResourceForm
ResourceForm filled with updated values.
Expand source code
def update(self, form: ResourceForm) -> dict:
        """
        Updates the metadata of the resource using the supplied ResourceForm.

        Parameters
        ----------
        form : ResourceForm
                ResourceForm filled with updated values.
        """

        if type(form) is ResourceForm:
                form = form.generate()
        elif type(form) is not dict:
                raise TypeError("")

        uri = self.client.uri("Resources", "Resource", self.id)
        response = self.client.post(uri, data = form)
        if response.ok: self.data = response.json()
        return response
def upload(self, key: str, file, metadata=None, callback: Callable[[int]] = None) ‑> None

Uploads a file-like object to a resource on the Coscine server

Parameters

key : str
filename of the file-like object.
file : object with read() attribute
Either open file handle or local file location path.
metadata : dict
File metadata. For rds-s3 this is optional, but recommended.
callback : Callable[int]
Optional callback called during chunk uploads.
Expand source code
def upload(self, key: str, file, metadata = None, \
                                callback: Callable[[int]] = None) -> None:
        """
        Uploads a file-like object to a resource on the Coscine server

        Parameters
        ----------
        key : str
                filename of the file-like object.
        file : object with read() attribute
                        Either open file handle or local file location path.
        metadata : dict
                File metadata. For rds-s3 this is optional, but recommended.
        callback : Callable[int]
                Optional callback called during chunk uploads.
        """

        if hasattr(file, "read"):
                fd = file
                filename = "MEM"
        elif type(file) is str:
                fd = open(file, "rb")
                filename = file
        else:
                raise TypeError("Argument `file` has unexpected type!")

        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.logger, 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)
class ResourceForm (client: Client)
Expand source code
class ResourceForm(InputForm):
        """
        """

###############################################################################

        def __init__(self, client: Client) -> None:
                self._items = client.vocabularies.builtin("resource")
                super().__init__(client)
                vocabularies = {
                        "type": client.vocabularies.resource_types(True),
                        "applicationProfile": client.vocabularies.application_profiles(True),
                        "license": client.vocabularies.licenses(True),
                        "visibility": client.vocabularies.visibility(True),
                        "disciplines": client.vocabularies.disciplines(True)
                }
                for item in self._items:
                        if item.vocabulary:
                                self._vocabularies[item.path] = vocabularies[item.path]

###############################################################################

        def parse(self, data: dict) -> None:
                IGNORE = ["id", "pid", "fixedValues", "creator", "archived"]
                if data is None:
                        return
                for path, value in data.items():
                        if path in IGNORE: continue
                        key = self.name_of(path)
                        self.set_value(key, value, True)

###############################################################################

        def generate(self) -> dict:
                metadata = {}

                # Collect missing required fields
                missing = []

                # Set metadata
                for key, values in self.items(True):
                        
                        if not values:
                                if self.is_required(key):
                                        missing.append(key)
                                continue

                        properties = self.properties(key)
                        metadata[properties.path] = values if len(values) > 1 else values[0]

                # Check for missing required fields
                if len(missing) > 0:
                        if len(missing) == 1 and missing[0] == "resourceTypeOption" \
                                                                        and     metadata["type"] == "linked":
                                pass
                        else:
                                raise ValueError(missing)

                return metadata

Ancestors

  • InputForm
  • collections.abc.MutableMapping
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container
  • typing.Generic

Class variables

var client : Client

Inherited members

class S3info (data: dict)

Provides a simple pythonic interface to resource S3 access data

Expand source code
class S3info:
        """
        Provides a simple pythonic interface to resource S3 access data
        """

        read_access_key: str
        read_secret_key: str
        write_access_key: str
        write_secret_key: str
        endpoint: str
        bucket: str
        _data: dict

        def __init__(self, data: dict) -> None:
                self._data = data
                if "resourceTypeOption" in data and data["resourceTypeOption"]:
                        opt: dict = data["resourceTypeOption"]
                        self.read_access_key = opt.get("ReadAccessKey")
                        self.read_secret_key = opt.get("ReadSecretKey")
                        self.write_access_key = opt.get("WriteAccessKey")
                        self.write_secret_key = opt.get("WriteSecretKey")
                        self.endpoint = opt.get("Endpoint")
                        self.bucket = opt.get("BucketName")

Class variables

var bucket : str
var endpoint : str
var read_access_key : str
var read_secret_key : str
var write_access_key : str
var write_secret_key : str