# Copyright ETH 2018 - 2023 Zürich, Scientific IT Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import getpass
import hashlib
import os
import socket
import pybis
from .. import config as dm_config
from ..command_result import CommandException
from ..command_result import CommandResult
from ..utils import complete_openbis_config, OperationType
class OpenbisCommand(object):
""" Superclass for commands connecting to openBIS.
"""
def __init__(self, dm):
self.data_mgmt = dm
self.openbis = dm.openbis
self.git_wrapper = dm.git_wrapper
self.settings_resolver = dm.settings_resolver
self.config_dict = dm.settings_resolver.config_dict()
if self.openbis is None and dm.openbis_config.get('url') is not None:
self.openbis = pybis.Openbis(url=dm.openbis_config.get('url'),
verify_certificates=dm.openbis_config.get(
'verify_certificates'),
token=dm.openbis_config.get('token'),
use_cache=dm.openbis_config.get('use_cache'),
allow_http_but_do_not_use_this_in_production_and_only_within_safe_networks=dm.openbis_config.get(
'allow_http_but_do_not_use_this_in_production_and_only_within_safe_networks'),
)
if self.user() is not None:
result = self.login()
if result.failure():
raise CommandException(result)
def external_dms_id(self):
return self.config_dict['repository']['external_dms_id']
def set_external_dms_id(self, value):
self.config_dict['repository']['external_dms_id'] = value
def repository_id(self):
return self.config_dict['repository']['id']
def set_repository_id(self, value):
self.config_dict['repository']['id'] = value
def data_set_id(self):
return self.config_dict['repository']['data_set_id']
def set_data_set_id(self, value):
self.config_dict['repository']['data_set_id'] = value
def data_set_type(self):
return self.config_dict['data_set']['type']
def set_data_set_type(self, value):
self.config_dict['data_set']['type'] = value
def data_set_properties(self):
return self.config_dict['data_set']['properties']
def set_data_set_properties(self, value):
self.config_dict['data_set']['properties'] = value
def object_id(self):
return self.config_dict['object']['id']
def object_permId(self):
return self.config_dict['object']['permId']
def set_object_id(self, value):
self.config_dict['object']['id'] = value
def collection_id(self):
return self.config_dict['collection']['id']
def collection_permId(self):
return self.config_dict['collection']['permId']
def set_collection_id(self, value):
self.config_dict['collection']['id'] = value
def user(self):
return self.config_dict['config']['user']
def set_user(self, value):
self.config_dict['config']['user'] = value
def hostname(self):
return self.config_dict['config']['hostname']
def set_hostname(self, value):
self.config_dict['config']['hostname'] = value
def fileservice_url(self):
return self.config_dict['config']['fileservice_url']
def set_fileservice_url(self, value):
self.config_dict['config']['fileservice_url'] = value
def git_annex_hash_as_checksum(self):
return self.config_dict['config']['git_annex_hash_as_checksum']
def set_git_annex_hash_as_checksum(self, value):
self.config_dict['config']['git_annex_hash_as_checksum'] = value
def openbis_url(self):
return self.config_dict['config']['openbis_url']
def set_openbis_url(self, value):
self.config_dict['config']['openbis_url'] = value
def openbis_token(self):
return self.config_dict['config']['openbis_token']
def set_openbis_token(self, value):
self.config_dict['config']['openbis_token'] = value
def log(self, message):
command = type(self).__name__
self.data_mgmt.log.log(command, message)
def prepare_run(self):
result = self.check_configuration()
if result.failure():
return result
result = self.login()
if result.failure():
return result
return CommandResult(returncode=0, output="")
def check_configuration(self):
""" overwrite in subclass """
return CommandResult(returncode=0, output="")
def login(self):
""" Restore session token if available. """
if 'config' in self.config_dict.keys():
if 'openbis_token' in self.config_dict['config'].keys() and \
self.config_dict['config']['openbis_token'] is not None:
self.openbis.set_token(self.config_dict['config']['openbis_token'], True)
""" Checks for valid session and asks user for password
if login is needed. """
user = self.user()
if self.openbis.is_session_active():
if self.openbis.token.startswith(user) or self.openbis.token.startswith('$pat-' + user):
return CommandResult(returncode=0, output="")
else:
self.openbis.logout()
if self.data_mgmt.login == False:
return CommandResult(returncode=-1, output="No active session.")
passwd = getpass.getpass("Password for {}:".format(user))
try:
self.openbis.login(user, passwd, save_token=True)
except ValueError:
msg = "Could not log into openbis {}".format(self.openbis_url())
return CommandResult(returncode=-1, output=msg)
return CommandResult(returncode=0, output='')
def prepare_external_dms(self):
# If there is no external data management system, create one.
result = self.get_or_create_external_data_management_system()
if result.failure():
return result
external_dms = result.output
self.settings_resolver.repository.set_value_for_parameter(
'external_dms_id', external_dms.code, 'local')
self.set_external_dms_id(external_dms.code)
return result
def generate_external_data_management_system_code(self, user, hostname, edms_path):
path_hash = hashlib.sha1(edms_path.encode("utf-8")).hexdigest()[0:8]
return "{}-{}-{}".format(user, hostname, path_hash).upper()
def get_or_create_external_data_management_system(self):
external_dms_id = self.external_dms_id()
user = self.user()
hostname = self.determine_hostname()
result = self.git_wrapper.git_top_level_path()
if result.failure():
return result
edms_path, path_name = os.path.split(result.output)
if external_dms_id is None:
external_dms_id = self.generate_external_data_management_system_code(
user, hostname, edms_path)
try:
external_dms = self.openbis.get_external_data_management_system(
external_dms_id.upper())
except ValueError:
# external dms does not exist - create it
try:
external_dms = self.openbis.create_external_data_management_system(external_dms_id,
external_dms_id,
"{}:/{}".format(
hostname,
edms_path))
except Exception as error:
return CommandResult(returncode=-1, output=str(error))
return CommandResult(returncode=0, output=external_dms)
def determine_hostname(self):
""" Returns globally defined hostname if available.
Otherwies, lets the user choose one and stores that globally. """
# from global config
hostname = self.hostname()
if hostname is not None:
return hostname
# ask user
hostname = self.ask_for_hostname(socket.gethostname())
# store
self.data_mgmt.config('config', True, False, OperationType.SET, prop='hostname',
value=hostname)
return hostname
def ask_for_hostname(self, hostname):
""" Asks the user to confirm the suggestes hostname or choose a custom one. """
hostname_input = input(
'Enter hostname (empty to confirm \'' + str(hostname) + '\'): ')
if hostname_input:
return hostname_input
else:
return hostname
def path(self):
result = self.git_wrapper.git_top_level_path()
if result.failure():
return result
return result.output
def load_global_config(self, dm):
"""
Use global config only.
"""
resolver = dm_config.SettingsResolver()
config = {}
complete_openbis_config(config, resolver, False)
dm.openbis_config = config
class ContentCopySelector(object):
""" In case a command needs information from a content copy, this class
asks the user to pick one if there are multiple. """
def __init__(self, data_set, content_copy_index=None, get_index=False):
self.data_set = data_set
self.content_copy_index = content_copy_index
self.get_index = get_index
def select(self):
content_copy_index = self.select_index()
if self.get_index == True:
return content_copy_index
else:
return self.data_set.data['linkedData']['contentCopies'][content_copy_index]
def select_index(self):
if self.data_set.data['kind'] != 'LINK':
raise ValueError('Data set is of type ' +
self.data_set.data['kind'] + ' but should be LINK.')
content_copies = self.data_set.data['linkedData']['contentCopies']
if len(content_copies) == 0:
raise ValueError("Data set has no content copies.")
elif len(content_copies) == 1:
return 0
else:
return self.select_content_copy_index(content_copies)
def select_content_copy_index(self, content_copies):
if self.content_copy_index is not None:
# use provided content_copy_index
if self.content_copy_index >= 0 and self.content_copy_index < len(content_copies):
return self.content_copy_index
else:
raise ValueError("Invalid content copy index.")
else:
# ask user
while True:
print('From which location should the files be copied?')
for i, content_copy in enumerate(content_copies):
host = content_copy['externalDms']['address'].split(":")[0]
path = content_copy['path']
print(" {}) {}:{}".format(i, host, path))
copy_index_string = input('> ')
if copy_index_string.isdigit():
copy_index_int = int(copy_index_string)
if copy_index_int >= 0 and copy_index_int < len(content_copies):
return copy_index_int