"""API for interacting with a remote Alyx instance through REST.
The AlyxClient class contains methods for making remote Alyx REST queries and downloading remote
files through Alyx.
Examples
--------
>>> alyx = AlyxClient(
... username='test_user', password='TapetesBloc18',
... base_url='https://test.alyx.internationalbrainlab.org')
List subjects
>>> subjects = alyx.rest('subjects', 'list')
Create a subject
>>> record = {
... 'nickname': nickname,
... 'responsible_user': 'olivier',
... 'birth_date': '2019-06-15',
... 'death_date': None,
... 'lab': 'cortexlab',
... }
>>> new_subj = alyx.rest('subjects', 'create', data=record)
Download a remote file, given a local path
>>> url = 'zadorlab/Subjects/flowers/2018-07-13/1/channels.probe.npy'
>>> local_path = alyx.download_file(url, target_dir='zadorlab/Subjects/flowers/2018-07-13/1/')
"""
import json
import logging
import math
import re
import functools
import urllib.request
from urllib.error import HTTPError
import urllib.parse
from collections.abc import Mapping
from typing import Optional
from datetime import datetime, timedelta
from pathlib import Path
import warnings
import hashlib
import zipfile
import tempfile
from getpass import getpass
from contextlib import contextmanager
import requests
from tqdm import tqdm
from pprint import pprint
import one.params
from iblutil.io import hashfile
from iblutil.io.params import set_hidden
from one.util import ensure_list
import concurrent.futures
_logger = logging.getLogger(__name__)
def _cache_response(method):
"""Decorator for the generic request method.
Caches the result of the query and on subsequent calls, returns cache instead of hitting the
database.
Parameters
----------
method : function
Function to wrap (i.e. AlyxClient._generic_request).
Returns
-------
function
Handle to wrapped method.
"""
@functools.wraps(method)
def wrapper_decorator(alyx_client, *args, expires=None, clobber=False, **kwargs):
"""
REST caching wrapper.
Parameters
----------
alyx_client : AlyxClient
An instance of the AlyxClient class.
args : any
Positional arguments for applying to wrapped function.
expires : bool
An optional timedelta for how long cached response is valid. If True, the cached
response will not be used on subsequent calls. If None, the default expiry is applied.
clobber : bool
If True any existing cached response is overwritten.
**kwargs
Keyword arguments for applying to wrapped function.
Returns
-------
dict
The REST response JSON either from cached file or directly from remote.
"""
expires = expires or alyx_client.default_expiry
mode = (alyx_client.cache_mode or '').lower()
if args[0].__name__ != mode and mode != '*':
return method(alyx_client, *args, **kwargs)
# Check cache
rest_cache = alyx_client.cache_dir.joinpath('.rest')
sha1 = hashlib.sha1()
sha1.update(bytes(args[1], 'utf-8'))
name = sha1.hexdigest()
# Reversible but length may exceed 255 chars
# name = base64.urlsafe_b64encode(args[2].encode('UTF-8')).decode('UTF-8')
files = list(rest_cache.glob(name))
cached = None
if len(files) == 1 and not clobber:
_logger.debug('loading REST response from cache')
with open(files[0], 'r') as f:
cached, when = json.load(f)
if datetime.fromisoformat(when) > datetime.now():
return cached
try:
response = method(alyx_client, *args, **kwargs)
except requests.exceptions.ConnectionError as ex:
if cached and not clobber:
warnings.warn('Failed to connect, returning cached response', RuntimeWarning)
return cached
raise ex # No cache and can't connect to database; re-raise
# Save response into cache
if not rest_cache.exists():
rest_cache.mkdir(parents=True)
rest_cache = set_hidden(rest_cache, True)
_logger.debug('caching REST response')
expiry_datetime = datetime.now() + (timedelta() if expires is True else expires)
with open(rest_cache / name, 'w') as f:
json.dump((response, expiry_datetime.isoformat()), f)
return response
return wrapper_decorator
[docs]
@contextmanager
def no_cache(ac=None):
"""Temporarily turn off the REST cache for a given Alyx instance.
This function is particularly useful when calling ONE methods in remote mode.
Parameters
----------
ac : AlyxClient
An instance of the AlyxClient to modify. If None, the a new object is instantiated
Returns
-------
AlyxClient
The instance of Alyx with cache disabled
Examples
--------
>>> from one.api import ONE
>>> with no_cache(ONE().alyx):
... eids = ONE().search(subject='foobar', query_type='remote')
"""
ac = ac or AlyxClient()
cache_mode = ac.cache_mode
ac.cache_mode = None
try:
yield ac
finally:
ac.cache_mode = cache_mode
class _PaginatedResponse(Mapping):
"""
This class allows to emulate a list from a paginated response.
Provides cache functionality.
Examples
--------
>>> r = _PaginatedResponse(client, response)
"""
def __init__(self, alyx, rep, cache_args=None):
"""
A paginated response cache object
Parameters
----------
alyx : AlyxClient
An instance of an AlyxClient associated with the REST response
rep : dict
A paginated REST response JSON dictionary
cache_args : dict
A dict of kwargs to pass to _cache_response decorator upon subsequent requests
"""
self.alyx = alyx
self.count = rep['count']
self.limit = len(rep['results'])
self._cache_args = cache_args or {}
# store URL without pagination query params
self.query = rep['next']
# init the cache, list with None with count size
self._cache = [None] * self.count
# fill the cache with results of the query
for i in range(self.limit):
self._cache[i] = rep['results'][i]
def __len__(self):
return self.count
def __getitem__(self, item):
if isinstance(item, slice):
while None in self._cache[item]:
self.populate(item.start + self._cache[item].index(None))
elif self._cache[item] is None:
self.populate(item)
return self._cache[item]
def populate(self, idx):
offset = self.limit * math.floor(idx / self.limit)
query = update_url_params(self.query, {'limit': self.limit, 'offset': offset})
res = self.alyx._generic_request(requests.get, query, **self._cache_args)
if self.count != res['count']:
warnings.warn(
f'remote results for {urllib.parse.urlsplit(query).path} endpoint changed; '
f'results may be inconsistent', RuntimeWarning)
for i, r in enumerate(res['results'][:self.count - offset]):
self._cache[i + offset] = res['results'][i]
def __iter__(self):
for i in range(self.count):
yield self.__getitem__(i)
[docs]
def update_url_params(url: str, params: dict) -> str:
"""Add/update the query parameters of a URL and make url safe
Parameters
----------
url : str
A URL string with which to update the query parameters
params : dict
A dict of new parameters. For multiple values for the same query, use a list (see example)
Returns
-------
str
A new URL with said parameters updated
Examples
-------
>>> update_url_params('website.com/?q=', {'pg': 5})
'website.com/?pg=5'
>>> update_url_params('website.com?q=xxx', {'pg': 5, 'foo': ['bar', 'baz']})
'website.com?q=xxx&pg=5&foo=bar&foo=baz'
"""
# Remove percent-encoding
url = urllib.parse.unquote(url)
parsed_url = urllib.parse.urlsplit(url)
# Extract URL query arguments and convert to dict
parsed_get_args = urllib.parse.parse_qs(parsed_url.query, keep_blank_values=False)
# Merge URL arguments dict with new params
parsed_get_args.update(params)
# Convert back to query string
encoded_get_args = urllib.parse.urlencode(parsed_get_args, doseq=True)
# Update parser and convert to full URL str
return parsed_url._replace(query=encoded_get_args).geturl()
[docs]
def http_download_file_list(links_to_file_list, **kwargs):
"""
Downloads a list of files from a remote HTTP server from a list of links.
Generates up to 4 separate threads to handle downloads.
Same options behaviour as http_download_file.
Parameters
----------
links_to_file_list : list
List of http links to files.
**kwargs
Optional arguments to pass to http_download_file.
Returns
-------
list of pathlib.Path
A list of the local full path of the downloaded files.
"""
links_to_file_list = list(links_to_file_list) # In case generator was passed
n_threads = 4 # Max number of threads
outputs = []
target_dir = kwargs.pop('target_dir', None)
# Ensure target dir the length of url list
if target_dir is None or isinstance(target_dir, (str, Path)):
target_dir = [target_dir] * len(links_to_file_list)
assert len(target_dir) == len(links_to_file_list)
# using with statement to ensure threads are cleaned up promptly
zipped = zip(links_to_file_list, target_dir)
with concurrent.futures.ThreadPoolExecutor(max_workers=n_threads) as executor:
# Multithreading load operations
futures = [executor.submit(
http_download_file, link, target_dir=target, **kwargs) for link, target in zipped]
zip(links_to_file_list, ensure_list(kwargs.pop('target_dir', None)))
# TODO Reintroduce variable timeout value based on file size and download speed of 5 Mb/s?
# timeout = reduce(lambda x, y: x + (y.get('file_size', 0) or 0), dsets, 0) / 625000 ?
concurrent.futures.wait(futures, timeout=None)
# build return list
for future in futures:
outputs.append(future.result())
# if returning md5, separate list of tuples into two lists: (files, md5)
return list(zip(*outputs)) if kwargs.get('return_md5', False) else outputs
[docs]
def http_download_file(full_link_to_file, chunks=None, *, clobber=False, silent=False,
username='', password='', target_dir='', return_md5=False, headers=None):
"""
Download a file from a remote HTTP server.
Parameters
----------
full_link_to_file : str
HTTP link to the file
chunks : tuple of ints
Chunks to download
clobber : bool
If True, force overwrite the existing file
silent : bool
If True, suppress download progress bar
username : str
User authentication for password protected file server
password : str
Password authentication for password protected file server
target_dir : str, pathlib.Path
Directory in which files are downloaded; defaults to user's Download directory
return_md5 : bool
If True an MD5 hash of the file is additionally returned
headers : list of dicts
Additional headers to add to the request (auth tokens etc.)
Returns
-------
pathlib.Path
The full file path of the downloaded file
"""
if not full_link_to_file:
return (None, None) if return_md5 else None
# makes sure special characters get encoded ('#' in file names for example)
surl = urllib.parse.urlsplit(full_link_to_file, allow_fragments=False)
full_link_to_file = surl._replace(path=urllib.parse.quote(surl.path)).geturl()
# default cache directory is the home dir
if not target_dir:
target_dir = Path.home().joinpath('Downloads')
# This should be the base url you wanted to access.
base_url, name = full_link_to_file.rsplit('/', 1)
file_name = Path(target_dir, name)
# do not overwrite an existing file unless specified
if not clobber and file_name.exists():
return (file_name, hashfile.md5(file_name)) if return_md5 else file_name
# Create a password manager
manager = urllib.request.HTTPPasswordMgrWithDefaultRealm()
if username and password:
manager.add_password(None, base_url, username, password)
# Create an authentication handler using the password manager
auth = urllib.request.HTTPBasicAuthHandler(manager)
# Create an opener that will replace the default urlopen method on further calls
opener = urllib.request.build_opener(auth)
urllib.request.install_opener(opener)
# Support for partial download.
req = urllib.request.Request(full_link_to_file)
if chunks is not None:
first_byte, n_bytes = chunks
req.add_header('Range', 'bytes=%d-%d' % (first_byte, first_byte + n_bytes - 1))
# add additional headers
if headers is not None:
for k in headers:
req.add_header(k, headers[k])
# Open the url and get the length
try:
u = urllib.request.urlopen(req)
except HTTPError as e:
_logger.error(f'{str(e)} {full_link_to_file}')
raise e
file_size = int(u.getheader('Content-length'))
if not silent:
print(f'Downloading: {file_name} Bytes: {file_size}')
block_sz = 8192 * 64 * 8
md5 = hashlib.md5()
f = open(file_name, 'wb')
with tqdm(total=file_size / 1024 / 1024, disable=silent) as pbar:
while True:
buffer = u.read(block_sz)
if not buffer:
break
f.write(buffer)
if return_md5:
md5.update(buffer)
pbar.update(len(buffer) / 1024 / 1024)
f.close()
return (file_name, md5.hexdigest()) if return_md5 else file_name
[docs]
def file_record_to_url(file_records) -> list:
"""
Translate a Json dictionary to an usable http url for downloading files.
Parameters
----------
file_records : dict
JSON containing a 'data_url' field
Returns
-------
list of str
A list of full data urls
"""
urls = []
for fr in file_records:
if fr['data_url'] is not None:
urls.append(fr['data_url'])
return urls
[docs]
def dataset_record_to_url(dataset_record) -> list:
"""
Extracts a list of files urls from a list of dataset queries.
Parameters
----------
dataset_record : list, dict
Dataset JSON from a REST request
Returns
-------
list of str
A list of file urls corresponding to the datasets records
"""
urls = []
if isinstance(dataset_record, dict):
dataset_record = [dataset_record]
for ds in dataset_record:
urls += file_record_to_url(ds['file_records'])
return urls
[docs]
class AlyxClient:
"""
Class that implements simple GET/POST wrappers for the Alyx REST API.
See https://openalyx.internationalbrainlab.org/docs
"""
_token = None
_headers = {} # Headers for REST requests only
user = None
"""str: The Alyx username."""
base_url = None
"""str: The Alyx database URL."""
def __init__(self, base_url=None, username=None, password=None,
cache_dir=None, silent=False, cache_rest='GET'):
"""
Create a client instance that allows to GET and POST to the Alyx server.
For One, constructor attempts to authenticate with credentials in params.py.
For standalone cases, AlyxClient(username='', password='', base_url='').
Parameters
----------
base_url : str
Alyx server address, including port and protocol.
username : str
Alyx database user.
password : str
Alyx database password.
cache_dir : str, pathlib.Path
The default root download location.
silent : bool
If true, user prompts and progress bars are suppressed.
cache_rest : str, None
Which type of http method to apply cache to; if '*', all requests are cached.
stay_logged_in : bool
If true, auth token is cached.
"""
self.silent = silent
self._par = one.params.get(client=base_url, silent=self.silent, username=username)
self.base_url = base_url or self._par.ALYX_URL
self._par = self._par.set('CACHE_DIR', cache_dir or self._par.CACHE_DIR)
if username or password:
self.authenticate(username, password)
self._rest_schemes = None
# the mixed accept application may cause errors sometimes, only necessary for the docs
self._headers = {**self._headers, 'Accept': 'application/json'}
# REST cache parameters
# The default length of time that cache file is valid for,
# The default expiry is overridden by the `expires` kwarg. If False, the caching is
# turned off.
self.default_expiry = timedelta(days=1)
self.cache_mode = cache_rest
self._obj_id = id(self)
@property
def rest_schemes(self):
"""dict: The REST endpoints and their parameters."""
# Delayed fetch of rest schemes speeds up instantiation
if not self._rest_schemes:
self._rest_schemes = self.get('/docs', expires=timedelta(weeks=1))
return self._rest_schemes
@property
def cache_dir(self):
"""pathlib.Path: The location of the downloaded file cache."""
return Path(self._par.CACHE_DIR)
@cache_dir.setter
def cache_dir(self, cache_dir):
cache_dir = Path(cache_dir)
cache_dir.mkdir(parents=True, exist_ok=True)
self._par = self._par.set('CACHE_DIR', cache_dir)
@property
def is_logged_in(self):
"""bool: Check if user logged into Alyx database; True if user is authenticated."""
return bool(self.user and self._token and 'Authorization' in self._headers)
[docs]
def list_endpoints(self):
"""
Return a list of available REST endpoints.
Returns
-------
List of REST endpoint strings.
"""
EXCLUDE = ('_type', '_meta', '', 'auth-token')
return sorted(x for x in self.rest_schemes.keys() if x not in EXCLUDE)
[docs]
def print_endpoint_info(self, endpoint, action=None):
"""
Print the available actions and query parameters for a given REST endpoint.
Parameters
----------
endpoint : str
An Alyx REST endpoint to query.
action : str
An optional action (e.g. 'list') to print. If None, all actions are printed.
Returns
-------
dict, list
A dictionary of endpoint query parameter details or a list of parameter details if
action is not None.
"""
rs = self.rest_schemes
if endpoint not in rs:
return print(f'Endpoint "{endpoint}" does not exist')
for _action in (rs[endpoint] if action is None else [action]):
doc = []
pprint(_action)
for f in rs[endpoint][_action]['fields']:
required = ' (required): ' if f.get('required', False) else ': '
doc.append(f'\t"{f["name"]}"{required}{f["schema"]["_type"]}'
f', {f["schema"]["description"]}')
doc.sort()
[print(d) for d in doc if '(required)' in d]
[print(d) for d in doc if '(required)' not in d]
return (rs[endpoint] if action is None else rs[endpoint][action]).copy()
@_cache_response
def _generic_request(self, reqfunction, rest_query, data=None, files=None):
if not self.is_logged_in:
self.authenticate(username=self.user)
# makes sure the base url is the one from the instance
rest_query = rest_query.replace(self.base_url, '')
if not rest_query.startswith('/'):
rest_query = '/' + rest_query
_logger.debug(f"{self.base_url + rest_query}, headers: {self._headers}")
headers = self._headers.copy()
if files is None:
data = json.dumps(data) if isinstance(data, dict) or isinstance(data, list) else data
headers['Content-Type'] = 'application/json'
if rest_query.startswith('/docs'):
# the mixed accept application may cause errors sometimes, only necessary for the docs
headers['Accept'] = 'application/coreapi+json'
r = reqfunction(self.base_url + rest_query,
stream=True, headers=headers, data=data, files=files)
if r and r.status_code in (200, 201):
return json.loads(r.text)
elif r and r.status_code == 204:
return
if r.status_code == 403 and '"Invalid token."' in r.text:
_logger.debug('Token invalid; Attempting to re-authenticate...')
# Log out in order to flush stale token. At this point we no longer have the password
# but if the user re-instantiates with a password arg it will request a new token.
username = self.user
if self.silent: # no need to log out otherwise; user will be prompted for password
self.logout()
self.authenticate(username=username, force=True)
return self._generic_request(reqfunction, rest_query, data=data, files=files)
else:
_logger.debug('Response text raw: ' + r.text)
try:
message = json.loads(r.text)
message.pop('status_code', None) # Get status code from response object instead
message = message.get('detail') or message # Get details if available
_logger.debug(message)
except json.decoder.JSONDecodeError:
message = r.text
raise requests.HTTPError(r.status_code, rest_query, message, response=r)
[docs]
def authenticate(self, username=None, password=None, cache_token=True, force=False):
"""
Gets a security token from the Alyx REST API to create requests headers.
Credentials are loaded via one.params.
Parameters
----------
username : str
Alyx username. If None, token not cached and not silent, user is prompted.
password : str
Alyx password. If None, token not cached and not silent, user is prompted.
cache_token : bool
If true, the token is cached for subsequent auto-logins.
force : bool
If true, any cached token is ignored.
"""
# Get username
if username is None:
username = getattr(self._par, 'ALYX_LOGIN', self.user)
if username is None and not self.silent:
username = input('Enter Alyx username:')
# Check if token cached
if not force and getattr(self._par, 'TOKEN', False) and username in self._par.TOKEN:
self._token = self._par.TOKEN[username]
self._headers = {
'Authorization': f'Token {list(self._token.values())[0]}',
'Accept': 'application/json'}
self.user = username
return
# Get password
if password is None:
password = getattr(self._par, 'ALYX_PWD', None)
if password is None and not self.silent:
password = getpass(f'Enter Alyx password for "{username}":')
try:
credentials = {'username': username, 'password': password}
rep = requests.post(self.base_url + '/auth-token', data=credentials)
except requests.exceptions.ConnectionError:
raise ConnectionError(
f"Can't connect to {self.base_url}.\n" +
"Check your internet connections and Alyx database firewall"
)
# Assign token or raise exception on auth error
if rep.ok:
self._token = rep.json()
assert list(self._token.keys()) == ['token']
else:
if rep.status_code == 400: # Auth error; re-raise with details
redacted = '*' * len(credentials['password']) if credentials['password'] else None
message = ('Alyx authentication failed with credentials: '
f'user = {credentials["username"]}, password = {redacted}')
raise requests.HTTPError(rep.status_code, rep.url, message, response=rep)
else:
rep.raise_for_status()
self._headers = {
'Authorization': 'Token {}'.format(list(self._token.values())[0]),
'Accept': 'application/json'}
if cache_token:
# Update saved pars
par = one.params.get(client=self.base_url, silent=True)
tokens = getattr(par, 'TOKEN', {})
tokens[username] = self._token
one.params.save(par.set('TOKEN', tokens), self.base_url)
# Update current pars
self._par = self._par.set('TOKEN', tokens)
self.user = username
if not self.silent:
print(f'Connected to {self.base_url} as user "{self.user}"')
[docs]
def logout(self):
"""Log out from Alyx.
Deletes the cached authentication token for the currently logged-in user.
"""
if not self.is_logged_in:
return
par = one.params.get(client=self.base_url, silent=True)
username = self.user
# Remove token from cache
if getattr(par, 'TOKEN', False) and username in par.TOKEN:
del par.TOKEN[username]
one.params.save(par, self.base_url)
# Remove token from local pars
if getattr(self._par, 'TOKEN', False) and username in self._par.TOKEN:
del self._par.TOKEN[username]
# Remove token from object
self.user = None
self._token = None
if self._headers and 'Authorization' in self._headers:
del self._headers['Authorization']
self.clear_rest_cache()
if not self.silent:
print(f'{username} logged out from {self.base_url}')
[docs]
def delete(self, rest_query):
"""
Sends a DELETE request to the Alyx server. Will raise an exception on any status_code
other than 200, 201.
Parameters
----------
rest_query : str
A REST query string either as a relative URL path complete URL.
Returns
-------
JSON interpreted dictionary from response.
Examples
--------
>>> AlyxClient.delete('/weighings/c617562d-c107-432e-a8ee-682c17f9e698')
>>> AlyxClient.delete(
... 'https://alyx.example.com/endpoint/c617562d-c107-432e-a8ee-682c17f9e698')
"""
return self._generic_request(requests.delete, rest_query)
[docs]
def download_file(self, url, **kwargs):
"""
Downloads a single file or list of files on the Alyx server from a
file record REST field URL.
Parameters
----------
url : str, list
Full url(s) of the file(s).
**kwargs
WebClient.http_download_file parameters.
Returns
-------
Local path(s) of downloaded file(s).
"""
if isinstance(url, str):
url = self._validate_file_url(url)
download_fcn = http_download_file
else:
url = (self._validate_file_url(x) for x in url)
download_fcn = http_download_file_list
pars = dict(
silent=kwargs.pop('silent', self.silent),
target_dir=kwargs.pop('target_dir', self._par.CACHE_DIR),
username=self._par.HTTP_DATA_SERVER_LOGIN,
password=self._par.HTTP_DATA_SERVER_PWD,
**kwargs
)
try:
files = download_fcn(url, **pars)
except HTTPError as ex:
if ex.code == 401:
ex.msg += (' - please check your HTTP_DATA_SERVER_LOGIN and '
'HTTP_DATA_SERVER_PWD ONE params, or username/password kwargs')
raise ex
return files
[docs]
def download_cache_tables(self, source=None, destination=None):
"""Downloads the Alyx cache tables to the local data cache directory.
Parameters
----------
source : str, pathlib.Path
The remote HTTP directory of the cache table (excluding the filename).
Default: AlyxClient.base_url.
destination : str, pathlib.Path
The target directory into to which the tables will be downloaded.
Returns
-------
List of parquet table file paths.
"""
# query the database for the latest cache; expires=None overrides cached response
if not self.is_logged_in:
self.authenticate()
source = str(source or f'{self.base_url}/cache.zip')
destination = destination or self.cache_dir
Path(destination).mkdir(exist_ok=True, parents=True)
headers = self._headers if source.startswith(self.base_url) else None
with tempfile.TemporaryDirectory(dir=destination) as tmp:
file = http_download_file(source,
headers=headers,
silent=self.silent,
target_dir=tmp,
clobber=True)
with zipfile.ZipFile(file, 'r') as zipped:
files = zipped.namelist()
zipped.extractall(destination)
return [Path(destination, table) for table in files]
def _validate_file_url(self, url):
"""Asserts that URL matches HTTP_DATA_SERVER parameter.
Currently only one remote HTTP server is supported for a given AlyxClient instance. If
the URL contains only the relative path part, the full URL is returned.
Parameters
----------
url : str
The full or partial URL to validate.
Returns
-------
The complete URL.
Examples
--------
>>> url = self._validate_file_url('https://webserver.net/path/to/file')
'https://webserver.net/path/to/file'
>>> url = self._validate_file_url('path/to/file')
'https://webserver.net/path/to/file'
"""
if url.startswith('http'): # A full URL
assert url.startswith(self._par.HTTP_DATA_SERVER), \
('remote protocol and/or hostname does not match HTTP_DATA_SERVER parameter:\n' +
f'"{url[:40]}..." should start with "{self._par.HTTP_DATA_SERVER}"')
elif not url.startswith(self._par.HTTP_DATA_SERVER):
url = self.rel_path2url(url)
return url
[docs]
def rel_path2url(self, path):
"""Given a relative file path, return the remote HTTP server URL.
It is expected that the remote HTTP server has the same file tree as the local system.
Parameters
----------
path : str, pathlib.Path
A relative ALF path (subject/date/number/etc.).
Returns
-------
A URL string.
"""
path = str(path).strip('/')
assert not path.startswith('http')
return f'{self._par.HTTP_DATA_SERVER}/{path}'
[docs]
def get(self, rest_query, **kwargs):
"""
Sends a GET request to the Alyx server. Will raise an exception on any status_code
other than 200, 201.
For the dictionary contents and list of endpoints, refer to:
https://openalyx.internationalbrainlab.org/docs
Parameters
----------
rest_query : str
A REST URL path, e.g. '/sessions?user=Hamish'.
**kwargs
Optional arguments to pass to _generic_request and _cache_response decorator.
Returns
-------
JSON interpreted dictionary from response.
"""
rep = self._generic_request(requests.get, rest_query, **kwargs)
if isinstance(rep, dict) and list(rep.keys()) == ['count', 'next', 'previous', 'results']:
if len(rep['results']) < rep['count']:
cache_args = {k: v for k, v in kwargs.items() if k in ('clobber', 'expires')}
rep = _PaginatedResponse(self, rep, cache_args)
else:
rep = rep['results']
return rep
[docs]
def patch(self, rest_query, data=None, files=None):
"""
Sends a PATCH request to the Alyx server.
For the dictionary contents, refer to:
https://openalyx.internationalbrainlab.org/docs
Parameters
----------
rest_query : str
The endpoint as full or relative URL.
data : dict, str
JSON encoded string or dictionary (c.f. requests).
files : dict, tuple
Files to attach (c.f. requests).
Returns
-------
Response object.
"""
return self._generic_request(requests.patch, rest_query, data=data, files=files)
[docs]
def post(self, rest_query, data=None, files=None):
"""
Sends a POST request to the Alyx server.
For the dictionary contents, refer to:
https://openalyx.internationalbrainlab.org/docs
Parameters
----------
rest_query : str
The endpoint as full or relative URL.
data : dict, str
JSON encoded string or dictionary (c.f. requests).
files : dict, tuple
Files to attach (c.f. requests).
Returns
-------
Response object.
"""
return self._generic_request(requests.post, rest_query, data=data, files=files)
[docs]
def put(self, rest_query, data=None, files=None):
"""
Sends a PUT request to the Alyx server.
For the dictionary contents, refer to:
https://openalyx.internationalbrainlab.org/docs
Parameters
----------
rest_query : str
The endpoint as full or relative URL.
data : dict, str
JSON encoded string or dictionary (c.f. requests).
files : dict, tuple
Files to attach (c.f. requests).
Returns
-------
requests.Response
Response object.
"""
return self._generic_request(requests.put, rest_query, data=data, files=files)
[docs]
def rest(self, url=None, action=None, id=None, data=None, files=None,
no_cache=False, **kwargs):
"""
alyx_client.rest(): lists endpoints
alyx_client.rest(endpoint): lists actions for endpoint
alyx_client.rest(endpoint, action): lists fields and URL
Example REST endpoint with all actions:
>>> client = AlyxClient()
>>> client.rest('subjects', 'list')
>>> client.rest('subjects', 'list', field_filter1='filterval')
>>> client.rest('subjects', 'create', data=sub_dict)
>>> client.rest('subjects', 'read', id='nickname')
>>> client.rest('subjects', 'update', id='nickname', data=sub_dict)
>>> client.rest('subjects', 'partial_update', id='nickname', data=sub_dict)
>>> client.rest('subjects', 'delete', id='nickname')
>>> client.rest('notes', 'create', data=nd, files={'image': open(image_file, 'rb')})
Parameters
----------
url : str
Endpoint name.
action : str
One of 'list', 'create', 'read', 'update', 'partial_update', 'delete'.
id : str
Lookup string for actions 'read', 'update', 'partial_update', and 'delete'.
data : dict
Data dictionary for actions 'update', 'partial_update' and 'create'.
files : dict, tuple
Option file(s) to upload.
no_cache : bool
If true the `list` and `read` actions are performed without returning the cache.
kwargs
Filters as per the Alyx REST documentation
cf. https://openalyx.internationalbrainlab.org/docs/
Returns
-------
list, dict
List of queried dicts ('list') or dict (other actions).
"""
# if endpoint is None, list available endpoints
if not url:
pprint(self.list_endpoints())
return
# remove beginning slash if any
if url.startswith('/'):
url = url[1:]
# and split to the next slash or question mark
endpoint = re.findall("^/*[^?/]*", url)[0].replace('/', '')
# make sure the queried endpoint exists, if not throw an informative error
if endpoint not in self.rest_schemes.keys():
av = [k for k in self.rest_schemes.keys() if not k.startswith('_') and k]
raise ValueError('REST endpoint "' + endpoint + '" does not exist. Available ' +
'endpoints are \n ' + '\n '.join(av))
endpoint_scheme = self.rest_schemes[endpoint]
# on a filter request, override the default action parameter
if '?' in url:
action = 'list'
# if action is None, list available actions for the required endpoint
if not action:
pprint(list(endpoint_scheme.keys()))
self.print_endpoint_info(endpoint)
return
# make sure the desired action exists, if not throw an informative error
if action not in endpoint_scheme:
raise ValueError('Action "' + action + '" for REST endpoint "' + endpoint + '" does ' +
'not exist. Available actions are: ' +
'\n ' + '\n '.join(endpoint_scheme.keys()))
# the actions below require an id in the URL, warn and help the user
if action in ['read', 'update', 'partial_update', 'delete'] and not id:
_logger.warning('REST action "' + action + '" requires an ID in the URL: ' +
endpoint_scheme[action]['url'])
return
# the actions below require a data dictionary, warn and help the user with fields list
data_required = 'fields' in endpoint_scheme[action]
if action in ['create', 'update', 'partial_update'] and data_required and not data:
pprint(endpoint_scheme[action]['fields'])
for act in endpoint_scheme[action]['fields']:
print("'" + act['name'] + "': ...,")
_logger.warning('REST action "' + action + '" requires a data dict with above keys')
return
# clobber=True means remote request always made, expires=True means response is not cached
cache_args = {'clobber': no_cache, 'expires': kwargs.pop('expires', False) or no_cache}
if action == 'list':
# list doesn't require id nor
assert endpoint_scheme[action]['action'] == 'get'
# add to url data if it is a string
if id:
# this is a special case of the list where we query a uuid. Usually read is better
if 'django' in kwargs.keys():
kwargs['django'] = kwargs['django'] + ','
else:
kwargs['django'] = ""
kwargs['django'] = f"{kwargs['django']}pk,{id}"
# otherwise, look for a dictionary of filter terms
if kwargs:
# Convert all lists in query params to comma separated list
query_params = {k: ','.join(map(str, ensure_list(v))) for k, v in kwargs.items()}
url = update_url_params(url, query_params)
return self.get('/' + url, **cache_args)
if not isinstance(id, str) and id is not None:
id = str(id) # e.g. may be uuid.UUID
if action == 'read':
assert endpoint_scheme[action]['action'] == 'get'
return self.get('/' + endpoint + '/' + id.split('/')[-1], **cache_args)
elif action == 'create':
assert endpoint_scheme[action]['action'] == 'post'
return self.post('/' + endpoint, data=data, files=files)
elif action == 'delete':
assert endpoint_scheme[action]['action'] == 'delete'
return self.delete('/' + endpoint + '/' + id.split('/')[-1])
elif action == 'partial_update':
assert endpoint_scheme[action]['action'] == 'patch'
return self.patch('/' + endpoint + '/' + id.split('/')[-1], data=data, files=files)
elif action == 'update':
assert endpoint_scheme[action]['action'] == 'put'
return self.put('/' + endpoint + '/' + id.split('/')[-1], data=data, files=files)
# JSON field interface convenience methods
def _check_inputs(self, endpoint: str) -> None:
# make sure the queried endpoint exists, if not throw an informative error
if endpoint not in self.rest_schemes.keys():
av = (k for k in self.rest_schemes.keys() if not k.startswith('_') and k)
raise ValueError('REST endpoint "' + endpoint + '" does not exist. Available ' +
'endpoints are \n ' + '\n '.join(av))
return
[docs]
def json_field_write(
self,
endpoint: str = None,
uuid: str = None,
field_name: str = None,
data: dict = None
) -> dict:
"""
Write data to JSON field. WILL NOT CHECK IF DATA EXISTS
NOTE: Destructive write!
Parameters
----------
endpoint : str, None
Valid alyx endpoint, defaults to None.
uuid : str, uuid.UUID, None
UUID or lookup name for endpoint.
field_name : str, None
Valid json field name, defaults to None.
data : dict, None
Data to write to json field, defaults to None.
Returns
-------
dict
Written data dict.
"""
self._check_inputs(endpoint)
# Prepare data to patch
patch_dict = {field_name: data}
# Upload new extended_qc to session
ret = self.rest(endpoint, "partial_update", id=uuid, data=patch_dict)
return ret[field_name]
[docs]
def json_field_update(
self,
endpoint: str = None,
uuid: str = None,
field_name: str = 'json',
data: dict = None
) -> dict:
"""
Non-destructive update of JSON field of endpoint for object.
Will update the field_name of the object with pk = uuid of given endpoint
If data has keys with the same name of existing keys it will squash the old
values (uses the dict.update() method).
Parameters
----------
endpoint : str
Alyx REST endpoint to hit.
uuid : str, uuid.UUID
UUID or lookup name of object.
field_name : str
Name of the json field.
data : dict
A dictionary with fields to be updated.
Returns
-------
dict
New patched json field contents as dict.
Examples
--------
>>> client = AlyxClient()
>>> client.json_field_update('sessions', 'eid_str', 'extended_qc', {'key': 'value'})
"""
self._check_inputs(endpoint)
# Load current json field contents
current = self.rest(endpoint, 'read', id=uuid)[field_name]
if current is None:
current = {}
if not isinstance(current, dict):
_logger.warning(
f'Current json field {field_name} does not contains a dict, aborting update'
)
return current
# Patch current dict with new data
current.update(data)
# Prepare data to patch
patch_dict = {field_name: current}
# Upload new extended_qc to session
ret = self.rest(endpoint, 'partial_update', id=uuid, data=patch_dict)
return ret[field_name]
[docs]
def json_field_remove_key(
self,
endpoint: str = None,
uuid: str = None,
field_name: str = 'json',
key: str = None
) -> Optional[dict]:
"""
Remove inputted key from JSON field dict and re-upload it to Alyx.
Needs endpoint, UUID and json field name.
Parameters
----------
endpoint : str
Endpoint to hit, defaults to None.
uuid : str, uuid.UUID
UUID or lookup name for endpoint.
field_name : str
JSON field name of object, defaults to None.
key : str
Key name of dictionary inside object, defaults to None.
Returns
-------
dict
New content of json field.
"""
self._check_inputs(endpoint)
current = self.rest(endpoint, 'read', id=uuid)[field_name]
# If no contents, cannot remove key, return
if current is None:
return current
# if contents are not dict, cannot remove key, return contents
if isinstance(current, str):
_logger.warning(f'Cannot remove key {key} content of json field is of type str')
return None
# If key not present in contents of json field cannot remove key, return contents
if current.get(key, None) is None:
_logger.warning(
f'{key}: Key not found in endpoint {endpoint} field {field_name}'
)
return current
_logger.info(f"Removing key from dict: '{key}'")
current.pop(key)
# Re-write contents without removed key
written = self.json_field_write(
endpoint=endpoint, uuid=uuid, field_name=field_name, data=current
)
return written
[docs]
def json_field_delete(
self, endpoint: str = None, uuid: str = None, field_name: str = None
) -> None:
"""
Set an entire field to null.
Note that this deletes all data from a given field. To delete only a single key from a
given JSON field, use `json_field_remove_key`.
Parameters
----------
endpoint : str
Endpoint to hit, defaults to None.
uuid : str, uuid.UUID
UUID or lookup name for endpoint.
field_name : str
The field name of object (e.g. 'json', 'name', 'extended_qc'), defaults to None.
Returns
-------
None
New content of json field.
"""
self._check_inputs(endpoint)
_ = self.rest(endpoint, 'partial_update', id=uuid, data={field_name: None})
return _[field_name]
[docs]
def clear_rest_cache(self):
"""Clear all REST response cache files for the base url."""
for file in self.cache_dir.joinpath('.rest').glob('*'):
file.unlink()