140 lines
6.2 KiB
Python
140 lines
6.2 KiB
Python
import atexit
|
|
from irods.session import iRODSSession
|
|
import irods
|
|
import storage.exceptions as StorageException
|
|
from storage.storage import BaseStorage
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# iRods support - https://pypi.org/project/python-irodsclient/
|
|
|
|
|
|
class iRODSStorage(BaseStorage):
|
|
|
|
TYPE = 'irods'
|
|
|
|
def __init__(self, url=None, username=None, password=None, source=None, destination=None, encryption_key=None, sender_name=None, sender_email=None):
|
|
# The iRODS zone is added to the url parameter. Use a '#' as seperator. This needs to be an Existing iRODS zone
|
|
# Ex: rdms-prod-icat.data.rug.nl#rug
|
|
(url, self.irods_zone) = url.split('#')
|
|
if destination:
|
|
destination = destination.strip('/')
|
|
|
|
super().__init__(url, username, password, source, destination, encryption_key, sender_name, sender_email)
|
|
|
|
# We need to clean up the iRODS session. Using atexit is the easiest way.
|
|
atexit.register(self.__close)
|
|
|
|
def __connect(self):
|
|
try:
|
|
assert(self.client)
|
|
except AttributeError:
|
|
# Connect to the iRODS server
|
|
self.client = None
|
|
try:
|
|
self.client = iRODSSession(host=self.url, port=1247, user=self.username, password=self.password, zone=self.irods_zone)
|
|
# Need to make a call to validate the authentication. So by checking the version, we know if we can authenticate...
|
|
logger.debug(f'iRODS {self.client.server_version} connection through *native* authentication')
|
|
except irods.exception.CAT_INVALID_AUTHENTICATION:
|
|
# Authentication scheme is not native (default), so we try PAM here
|
|
try:
|
|
self.client = iRODSSession(host=self.url, port=1247, user=self.username, password=self.password, zone=self.irods_zone, irods_authentication_scheme='pam')
|
|
logger.debug(f'iRODS {self.client.server_version} connection through *PAM* authentication')
|
|
except irods.exception.CAT_INVALID_AUTHENTICATION:
|
|
# Authentication scheme is not PAM either last try: GIS
|
|
try:
|
|
self.client = iRODSSession(host=self.url, port=1247, user=self.username, password=self.password, zone=self.irods_zone, irods_authentication_scheme='gis')
|
|
logger.debug(f'iRODS {self.client.server_version} connection through *GIS* authentication')
|
|
except irods.exception.CAT_INVALID_AUTHENTICATION:
|
|
pass
|
|
|
|
if self.client is None:
|
|
logger.error('Unable to login to the iRODS instance. Please check username and password combination!')
|
|
raise StorageException.InvalidAuthentication(self.username)
|
|
|
|
logger.info('Created iRODS connection')
|
|
|
|
def __close(self):
|
|
logger.debug('Closing iRODS storage connection and clean up')
|
|
self.client.cleanup()
|
|
|
|
def _file_exists_action(self, path):
|
|
self.__connect()
|
|
try:
|
|
self.client.data_objects.get(f'/{self.irods_zone}/home/{self.username}/{path}')
|
|
except irods.exception.DataObjectDoesNotExist:
|
|
logger.debug(f'File \'{path}\' does NOT exists on the iRODS server')
|
|
return False
|
|
except irods.exception.CollectionDoesNotExist:
|
|
logger.debug(f'Parent folder of file \'{path}\' does NOT exists on the iRODS server')
|
|
return False
|
|
|
|
return True
|
|
|
|
def _directory_exists_action(self, path):
|
|
self.__connect()
|
|
try:
|
|
self.client.collections.get(f'/{self.irods_zone}/home/{self.username}/{path}')
|
|
logger.debug(f'Folder \'{path}\' exists on the iRODS server')
|
|
except irods.exception.CollectionDoesNotExist:
|
|
logger.debug(f'Folder \'{path}\' does NOT exists on the iRODS server')
|
|
return False
|
|
|
|
return True
|
|
|
|
def _make_folder_action(self, path):
|
|
self.__connect()
|
|
try:
|
|
self.client.collections.create(f'/{self.irods_zone}/home/{self.username}/{path}')
|
|
except irods.exception.CollectionDoesNotExist:
|
|
logger.debug(f'Parent folder of file \'{path}\' does NOT exists on the iRODS server')
|
|
return False
|
|
|
|
return True
|
|
|
|
def _upload_file_action(self, source, destination):
|
|
self.__connect()
|
|
# The upload path consists of a zone, username and path
|
|
destination = f'/{self.irods_zone}/home/{self.username}/{destination}'
|
|
logger.debug(f'Uploading to file: \'{destination}\'')
|
|
try:
|
|
obj = self.client.data_objects.create(destination)
|
|
logger.debug(f'Created file: \'{destination}\'')
|
|
# Open 'both' files and copy 4K data each time.
|
|
with obj.open('w') as irods_file, open(source, 'rb') as source_file_binary:
|
|
while True:
|
|
buf = source_file_binary.read(4096)
|
|
if buf:
|
|
irods_file.write(buf)
|
|
else:
|
|
break
|
|
|
|
obj.metadata.add('source', f'Upload from VRE DataDropOff\n Added file: {destination} uploaded by: {self.sender_name}({self.sender_email})')
|
|
|
|
except irods.exception.OVERWRITE_WITHOUT_FORCE_FLAG:
|
|
logger.warning('The uploaded file already exists. So we did NOT upload the new file!')
|
|
return False
|
|
|
|
return True
|
|
|
|
def _download_file_action(self, source, destination):
|
|
self.__connect()
|
|
logger.debug(f'Downloading file: \'{source}\' to \'{destination}\'')
|
|
try:
|
|
obj = self.client.data_objects.get(f'/{self.irods_zone}/home/{self.username}/{source}')
|
|
# Open 'both' files and copy 4K data each time.
|
|
with obj.open('r') as irods_source_file, open(destination, 'wb') as local_destination_file:
|
|
while True:
|
|
buf = irods_source_file.read(4096)
|
|
if buf:
|
|
local_destination_file.write(buf)
|
|
else:
|
|
break
|
|
|
|
except irods.exception.DataObjectDoesNotExist:
|
|
logger.error(f'File: \'{source}\' does not exists on the iRODS server')
|
|
return False
|
|
|
|
return True
|