"""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/')
"""
from uuid import UUID
import abc
import json
import logging
import math
import os
import re
import functools
import urllib.request
from urllib.error import HTTPError
import urllib.parse
from collections.abc import Mapping
from typing import Optional, List
from inspect import unwrap
from datetime import datetime, timedelta, timezone
from pathlib import Path
from weakref import ReferenceType
import warnings
import hashlib
import zipfile
import tempfile
from getpass import getpass
from contextlib import contextmanager
from random import uniform
from time import sleep
from email.utils import parsedate_to_datetime
import requests
from tqdm import tqdm
from pprint import pprint
import one.params
from one import __version__
from iblutil.io import hashfile
from iblutil.io.params import set_hidden
from iblutil.util import ensure_list
import concurrent.futures
_logger = logging.getLogger(__name__)
N_THREADS = int(os.environ.get('ONE_HTTP_DL_THREADS', 4))
"""int: The number of download threads."""
class _JSONEncoder(json.JSONEncoder):
"""A JSON encoder that handles UUID objects."""
def default(self, o):
"""Cast UUID objects to str before serializing."""
if isinstance(o, UUID):
return str(o)
return super().default(o)
def _cache_response(method):
"""Decorator for the generic request method for caching REST responses.
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 '').casefold()
if args[0].__name__ != mode and mode != '*':
return method(alyx_client, *args, **kwargs)
# Check cache
rest_cache = alyx_client.rest_cache_dir
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, when = None, None
if len(files) == 1 and not clobber:
_logger.debug('loading REST response from cache')
try:
with open(files[0], 'r') as f:
cached, when = json.load(f)
except json.JSONDecodeError as e:
_logger.debug(f'corrupted cache file {name}: {e}')
files[0].unlink() # delete corrupted cache file
if cached and 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, cls=_JSONEncoder)
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):
"""Emulate a list from a paginated response.
Provides cache functionality.
Examples
--------
>>> r = _PaginatedResponse(client, response)
"""
def __init__(self, alyx, rep, cache_args=None):
"""Emulate a list from a paginated response.
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]
self._callbacks = set()
def add_callback(self, cb):
"""Add a callback function to use each time a new page is fetched.
The callback function will be called with the page results each time :meth:`populate`
is called.
Parameters
----------
cb : callable
A callable that takes the results of each paginated resonse.
"""
if not callable(cb):
raise TypeError(f'Expected type "callable", got "{type(cb)}" instead')
else:
self._callbacks.add(cb)
def __len__(self):
return self.count
def __getitem__(self, item):
if isinstance(item, slice):
while None in self._cache[item]:
# If slice start index is -ve, convert to +ve index
i = self.count + item.start if item.start < 0 else item.start
self.populate(i + self._cache[item].index(None))
elif self._cache[item] is None:
# A -ve index is resolved against the count at the time of the request. If the remote
# count changed, populate re-syncs it and the record may now be on a different page,
# so resolve the index again and retry once.
for _ in range(2):
# If index is -ve, convert to +ve
self.populate(self.count + item if item < 0 else item)
if self._cache[item] is not None:
break
return self._cache[item]
def _resize(self, count):
"""Resize the cache to match a changed remote count.
Records are truncated from the end when the remote count shrinks, and None placeholders
appended when it grows. Note that cached records may be shifted relative to the remote
list, hence the warning raised by :meth:`populate`.
Parameters
----------
count : int
The new total number of remote records.
"""
if count < self.count:
del self._cache[count:]
else:
self._cache.extend([None] * (count - self.count))
self.count = count
def populate(self, idx):
"""Populate response cache with new page of results.
Fetches the specific page of results containing the index passed and populates
stores the results in the :prop:`_cache` property.
Parameters
----------
idx : int
The index of a given record to fetch.
"""
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)
self._resize(res['count'])
for i, r in enumerate(res['results'][:self.count - offset]):
self._cache[i + offset] = r
# Notify callbacks
pending_removal = []
for callback in self._callbacks:
# Handle weak reference callbacks first
if isinstance(callback, ReferenceType):
wf = callback
if (callback := wf()) is None:
pending_removal.append(wf)
continue
callback(res['results'])
for wf in pending_removal:
self._callbacks.discard(wf)
# When cache is complete, clear our callbacks
if all(reversed(self._cache)):
self._callbacks.clear()
def __iter__(self):
for i in range(self.count):
yield self.__getitem__(i)
def _parse_lifecycle_date(value: str) -> Optional[datetime]:
"""Parse the value of a Deprecation or Sunset response header.
Parameters
----------
value : str
Either a structured field date (RFC 9651), e.g. '@1688169599', or an HTTP-date
(RFC 9110), e.g. 'Sat, 31 Dec 2025 23:59:59 GMT'.
Returns
-------
datetime, None
The parsed timezone-aware date, or None if the value could not be parsed (e.g. the
legacy 'true' value of earlier Deprecation header drafts).
"""
value = value.strip()
try:
if value.startswith('@'):
return datetime.fromtimestamp(int(value[1:]), tz=timezone.utc)
return parsedate_to_datetime(value)
except (ValueError, TypeError, OverflowError):
return None
[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):
"""Download 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
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 : dict
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()
# The default target is ~/Downloads, which a container or CI runner may not have.
file_name.parent.mkdir(parents=True, exist_ok=True)
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:
"""Extract 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 RestScheme(abc.ABC):
"""Alyx REST scheme documentation."""
EXCLUDE = ('_type', '_meta', '', 'auth-token', 'api')
"""list: A list of endpoint names to exclude from listings."""
def __init__(self, rest_scheme: dict):
self._rest_scheme = rest_scheme
def _validate(self, endpoint: str, action: str = None) -> bool:
"""Validate that an endpoint and action exist.
This is used by `print_endpoint_info` to print a message if the endpoint or action do not
exist.
Parameters
----------
endpoint : str
The endpoint name.
action : str
The action name.
Returns
-------
bool
True if the endpoint and action exist, False otherwise.
"""
if endpoint not in self.endpoints:
print(f'Endpoint "{endpoint}" does not exist')
return False
if action is not None and action not in self.actions(endpoint):
print(
f'Endpoint "{endpoint}" does not have action "{action}": '
f'available actions: {", ".join(self.actions(endpoint))}'
)
return False
return True
[docs]
@abc.abstractmethod
def endpoints(self) -> list:
"""Return the list of available endpoints.
Returns
-------
list
List of available endpoints
"""
[docs]
@abc.abstractmethod
def actions(self, endpoint: str, *args) -> list:
"""Return a list of available actions for a given endpoint.
Parameters
----------
endpoint : str
The endpoint name to find actions for
Returns
-------
list
List of available actions for the endpoint
"""
[docs]
@abc.abstractmethod
def fields(self, endpoint: str, action: str) -> List[dict]:
"""Return a list of fields for a given endpoint/action combination.
Parameters
----------
endpoint : str
The endpoint name to find fields and parameters for
action : str
The Django REST action:
'list', 'read', 'create', 'update', 'delete', 'partial_update'
Returns
-------
list
List of dictionaries containing fields for the endpoint
"""
[docs]
def field_names(self, endpoint: str, action: str) -> List[str]:
"""Return a list of fields for a given endpoint/action combination.
Parameters
----------
endpoint : str
The endpoint name to find fields and parameters for
action : str
The Django REST action:
'list', 'read', 'create', 'update', 'delete', 'partial_update'
Returns
-------
list
List of strings containing fields names for the endpoint
"""
return [field['name'] for field in self.fields(endpoint, action)]
[docs]
@abc.abstractmethod
def print_endpoint_info(self, endpoint: str, action: str = None) -> None:
"""Prints relevant information about an endpoint and its actions.
Parameters
----------
endpoint : str
The endpoint name to find actions for
action : str
The Django REST action to perform:
'list', 'read', 'create', 'update', 'delete', 'partial_update'
Returns
-------
None
"""
[docs]
def url(self, endpoint: str, action: str):
"""Returns the url for a given endpoint/action combination.
Parameters
----------
endpoint : str
The endpoint name to find actions for
action : str
The Django REST action to perform:
'list', 'read', 'create', 'update', 'delete', 'partial_update'
Returns
-------
str
url of the endpoint: for example /sessions/{id}
"""
[docs]
def validate_endpoint_action(func):
"""Decorator to validate endpoint and action before executing a method.
This decorator checks if the endpoint exists and, if an action is provided,
whether the action exists for that endpoint. If validation fails, it returns
None, otherwise it executes the decorated function.
"""
@functools.wraps(func)
def wrapper(self, endpoint, action=None, *args, **kwargs):
if not self._validate(endpoint, action):
return
return func(self, endpoint, action, *args, **kwargs)
return wrapper
[docs]
class RestSchemeCoreApi(RestScheme):
"""Legacy Alyx REST scheme documentation."""
backend = 'core_api'
"""str: The name of the backend that generates the REST API documentation."""
@property
def endpoints(self) -> list:
return sorted(x for x in self._rest_scheme.keys() if x not in self.EXCLUDE)
[docs]
@validate_endpoint_action
def actions(self, endpoint: str, *args) -> list:
return sorted(x for x in self._rest_scheme[endpoint].keys() if x not in self.EXCLUDE)
[docs]
@validate_endpoint_action
def url(self, endpoint: str, action: str) -> str:
return self._rest_scheme[endpoint][action]['url']
[docs]
@validate_endpoint_action
def fields(self, endpoint: str, action: str) -> list:
return self._rest_scheme[endpoint][action]['fields']
[docs]
@validate_endpoint_action
def print_endpoint_info(self, endpoint: str, action: str = None) -> None:
for _action in self._rest_scheme[endpoint] if action is None else [action]:
doc = []
pprint(_action)
for f in self._rest_scheme[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]
[docs]
class RestSchemeOpenApi(RestScheme):
"""OpenAPI v3 Alyx REST scheme documentation."""
backend = 'open_api_v3'
"""str: The name of the backend that generates the REST API documentation."""
@property
def endpoints(self) -> list:
endpoints = set()
for path in self._rest_scheme['paths'].keys():
# Extract the endpoint name (word between first and second slash,
# or after first slash if no second slash)
match = re.match(r'^/([^/]+)', path)
if match and match.group(1) not in self.EXCLUDE:
endpoints.add(match.group(1))
return sorted(list(endpoints))
@validate_endpoint_action
def _get_endpoint_actions(self, endpoint: str, *args) -> dict:
# Find all paths that start with this endpoint
endpoint_paths = []
for path in self._rest_scheme['paths'].keys():
# Match either /endpoint or /endpoint/{something}
if path == f'/{endpoint}' or path.startswith(f'/{endpoint}/'):
endpoint_paths.append(path)
# Extract available HTTP methods (actions) for these paths
actions = {}
for path in endpoint_paths:
for method in self._rest_scheme['paths'][path].keys():
# Convert HTTP methods to standard action names
if method == 'get':
# Determine if this is a list or read action based on path pattern
if '{' in path: # Path has a parameter, likely a read action
action = 'read'
else:
action = 'list'
elif method == 'post':
action = 'create'
elif method == 'put':
action = 'update'
elif method == 'patch':
action = 'partial_update'
elif method == 'delete':
action = 'delete'
actions[action] = path
return actions
@validate_endpoint_action
def _endpoint_action_info(self, endpoint: str, action: str) -> dict:
# Find all paths that start with this endpoint
endpoint_paths = []
for path in self._rest_scheme['paths'].keys():
# Match either /endpoint or /endpoint/{something}
if path == f'/{endpoint}' and not action == 'read' or path.startswith(f'/{endpoint}/'):
endpoint_paths.append(path)
# Extract available HTTP methods (actions) for these paths
rest2http = {
'list': 'get',
'read': 'get',
'create': 'post',
'update': 'put',
'partial_update': 'patch',
'delete': 'delete',
}
endpoint_method_info = {'fields': {}, 'description': '', 'parameters': [], 'url': ''}
for path in endpoint_paths:
operation = self._rest_scheme['paths'][path].get(rest2http[action], None)
if operation is not None:
endpoint_method_info['url'] = path
if 'requestBody' in operation:
schema = operation['requestBody']['content']['application/json']['schema'][
'$ref'
].split('/')[-1]
endpoint_method_info['fields'] = self._rest_scheme['components']['schemas'][
schema
]['properties']
if 'description' in operation:
endpoint_method_info['description'] = operation['description']
if 'parameters' in operation:
endpoint_method_info['parameters'] = operation['parameters']
break
return endpoint_method_info
[docs]
def fields(self, endpoint: str, action: str) -> list:
# in openapi there is a distinction between fields and parameters
params = self._endpoint_action_info(endpoint, action)['parameters']
fields = self._endpoint_action_info(endpoint, action)['fields']
fields = [{'name': k, 'schema': v} for k, v in fields.items()]
return params + fields
[docs]
@validate_endpoint_action
def actions(self, endpoint: str, *args) -> list:
actions = self._get_endpoint_actions(endpoint)
return sorted(list(actions.keys()))
[docs]
def url(self, endpoint: str, action: str) -> str:
return self._endpoint_action_info(endpoint, action)['url']
[docs]
@validate_endpoint_action
def print_endpoint_info(self, endpoint: str, action: str = None) -> None:
"""Print detailed information about an endpoint's actions and parameters.
Parameters
----------
endpoint : str
The endpoint name to display information for
action : str, optional
If provided, only show information for this specific action
Returns
-------
None
"""
actions = self._get_endpoint_actions(endpoint) if action is None else [action]
for action in actions:
endpoint_action_info = self._endpoint_action_info(endpoint, action)
print(action)
for par in endpoint_action_info.get('parameters', []):
print(f'\t{par["name"]}')
all_fields = endpoint_action_info.get('fields', {})
if isinstance(all_fields, dict):
for field, field_info in all_fields.items():
print(f'\t{field}: ({field_info.get("type", "")}), '
f'{field_info.get("description", "")}')
[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."""
max_retry_attempts = 8
"""int: The maximum number of retry attempts for REST requests in case of throttling."""
max_retry_wait_seconds = 300. # 5 minutes
"""float: The maximum number of seconds to wait between retry attempts for REST requests."""
token_verify_timeout = 10.
"""float: Seconds to wait when checking a user-supplied token is valid."""
def __init__(
self,
base_url=None,
username=None,
password=None,
cache_dir=None,
silent=False,
cache_rest='GET',
token=None,
):
"""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.
token : str, dict
An Alyx API token, as an alternative to a password. Accounts that sign in through
an identity provider have no password and must use one.
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 or token:
self.authenticate(username, password, token=token)
self._rest_schemes = None
# the mixed accept application may cause errors sometimes, only necessary for the docs
self._headers = {**self._headers, **self._base_headers}
# 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(minutes=5)
self.rest_cache_dir = self.cache_dir.joinpath('.rest')
self.cache_mode = cache_rest
self._obj_id = id(self)
# Used to track number of attempts for a given request, for retry logic in _generic_request
self._attempt_counter = 0
# Response header notices already logged, so that each is logged only once per client
self._logged_notices = set()
@property
def _base_headers(self):
"""dict: The headers every REST request carries, authentication aside."""
return {'Accept': 'application/json', 'ONE-API-Version': __version__}
@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)
@property
def rest_schemes(self) -> RestScheme:
"""dict: The REST endpoints and their parameters."""
# Delayed fetch of rest schemes speeds up instantiation
if not self._rest_schemes:
try:
raw_schema = self.get('/api/schema', expires=timedelta(weeks=1))
self._rest_schemes = RestSchemeOpenApi(raw_schema)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
raw_schema = self.get('/docs', expires=timedelta(weeks=1))
self._rest_schemes = RestSchemeCoreApi(raw_schema)
else:
raise e
return self._rest_schemes
[docs]
def list_endpoints(self):
endpoints = self.rest_schemes.endpoints
pprint(endpoints)
return endpoints
[docs]
def print_endpoint_info(self, endpoint: str, action: str = None):
self.rest_schemes.print_endpoint_info(endpoint, action)
def _log_response_headers(self, response: requests.Response, rest_query: str) -> None:
"""Log the Alyx-API-Notice, Deprecation (RFC 9745) and Sunset (RFC 8594) headers.
Each distinct notice is logged only once per client, so that paginated queries and
retries do not repeat it.
Parameters
----------
response : requests.Response
The response whose headers to check.
rest_query : str
The REST query that was made, used in the deprecation message.
"""
headers = response.headers
def notify(message, emit):
"""Pass the message to `emit` unless this client has already done so."""
if message not in self._logged_notices:
self._logged_notices.add(message)
emit(message)
def see(urls):
"""Return a ' See ...' suffix listing the unique, non-empty URLs."""
urls = list(dict.fromkeys(filter(None, urls)))
return (' See ' + ', '.join(urls)) if urls else ''
# Alyx-API-Notice: logged, or raised as a warning if prefixed with '!!!'
if notice := headers.get('Alyx-API-Notice'):
links = see(link.get('url') for link in response.links.values())
if notice.startswith('!!!'):
warn = functools.partial(warnings.warn, category=RuntimeWarning)
notify(notice[3:] + links, warn)
else:
notify('Alyx-API-Notice: ' + notice + links, _logger.info)
deprecation = headers.get('Deprecation')
sunset = headers.get('Sunset')
if not (deprecation or sunset):
return
path = urllib.parse.urlsplit(rest_query).path
now = datetime.now(timezone.utc)
parts = []
if deprecation:
if (date := _parse_lifecycle_date(deprecation)) is None:
parts.append(f'{path} is deprecated')
elif date <= now:
parts.append(f'{path} has been deprecated since {date:%Y-%m-%d}')
else:
parts.append(f'{path} will be deprecated on {date:%Y-%m-%d}')
if sunset:
prefix = 'and' if parts else path
if (date := _parse_lifecycle_date(sunset)) is None:
parts.append(f'{prefix} is scheduled for removal')
elif date <= now:
parts.append(f'{prefix} was scheduled for removal on {date:%Y-%m-%d}')
else:
parts.append(f'{prefix} is scheduled for removal on {date:%Y-%m-%d}')
# Both headers may be accompanied by a Link header pointing to further information
links = (response.links.get(rel, {}).get('url') for rel in ('deprecation', 'sunset'))
notify(' '.join(parts) + '.' + see(links), _logger.warning)
@_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:
to_json = functools.partial(json.dumps, cls=_JSONEncoder)
data = to_json(data) if isinstance(data, dict) or isinstance(data, list) else data
headers['Content-Type'] = 'application/json'
if rest_query.startswith('/docs'):
headers['Accept'] = 'application/coreapi+json'
if rest_query.startswith('/api/schema'):
# the docs are now served as openapi media, so we need to change the accept header
headers['Accept'] = 'application/vnd.oai.openapi+json'
headers['Accept-Version'] = '3.0'
r = reqfunction(
self.base_url + rest_query, stream=True, headers=headers, data=data, files=files
)
self._log_response_headers(r, rest_query)
if r and r.status_code in (200, 201, 204):
self._attempt_counter = 0 # reset attempt counter on successful request
return None if r.status_code == 204 else json.loads(r.text)
if r.status_code == 403 and '"Invalid token."' in r.text:
_logger.info('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 unwrap(self._generic_request)(
self, reqfunction, rest_query, data=data, files=files)
if r.status_code in (429, 503) and self._attempt_counter < self.max_retry_attempts:
# 429 - Too Many Requests
# 503 - Service Unavailable
MAX_JITTER_SECONDS = 60.0
retry_after_seconds = 0.0
if retry_after := r.headers.get('Retry-After'):
# Usually Retry-After is an integer number of seconds
# but theoretically may be an HTTP date
try:
retry_after_seconds = max(0.0, float(retry_after))
except ValueError:
dt = parsedate_to_datetime(retry_after)
retry_after_seconds = max(
0.0, (dt - datetime.now(dt.tzinfo)).total_seconds())
# Add jitter to avoid synchronized retries across clients
# Jitter window grows exponentially with each retry attempt, up to a maximum
jitter_window = min(MAX_JITTER_SECONDS, 2 ** self._attempt_counter)
delay_seconds = retry_after_seconds + uniform(0, jitter_window)
# If server returns an excessive retry-after, simply raise an error instead of waiting
if delay_seconds < self.max_retry_wait_seconds:
_msg = 'Rate limited' if r.status_code == 429 else 'Service unavailable'
n_attempts = self._attempt_counter + 1
_logger.warning(
'%s for query: %s. Retrying in %.2f seconds (attempt %d/%d).',
_msg, rest_query, delay_seconds, n_attempts, self.max_retry_attempts
)
sleep(delay_seconds)
self._attempt_counter += 1
return unwrap(self._generic_request)(
self, reqfunction, rest_query, data=data, files=files)
# If we exhausted our retries, or if the error is not a 403 or 429,
# raise an HTTPError with details from the response
self._attempt_counter = 0 # reset attempt counter after exhausting retries
_logger.debug('Response text raw: ' + r.text)
try:
message = json.loads(r.text)
if isinstance(message, dict):
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,
token=None):
"""Fetch token from the Alyx REST API for authenticating request headers.
Credentials are loaded via one.params.
A token may be given in place of a password. Accounts that sign in through an identity
provider have no password at all and cannot use /auth-token, so a token copied from the
database's /me page is the only credential they have. The token is cached exactly as one
fetched with a password would be, so this is needed once rather than every session.
Parameters
----------
username : str
Alyx username. If None, token not cached and not silent, user is prompted.
Not required, and not prompted for, when `token` is given: the database reports
whose token it is.
password : str
Alyx password. If None, token not cached and not silent, user is prompted for a
password or, failing that, a token.
cache_token : bool
If true, the token is cached for subsequent auto-logins.
force : bool
If true, any cached token is ignored.
token : str, dict
An Alyx API token, used instead of a password. Obtained from the database's /me
page. Supplying one forces re-authentication, as supplying a password does.
"""
# An empty login counts as absent: the defaults now ship a blank one so a public
# database prompts rather than assuming a shared account.
#
# None of this applies to a token, which names its own owner through /me. Falling back
# to a stored login would cache the token under a name that may not own it.
if not username and token is None:
username = getattr(self._par, 'ALYX_LOGIN', None) or self.user
if not username and token is None and not self.silent:
username = input('Enter Alyx username:') or None
if not username and token is None:
# A token can name its own owner; nothing else can, so there is no point going on.
raise ValueError(
'No Alyx username. Pass one to ONE, or store it with '
'ONE.setup(username=<username>). Accounts on the public database are created '
'at https://openalyx.internationalbrainlab.org/signup')
# If the user passes in a credential, force re-authentication even if token cached
if password is not None or token is not None:
if not force:
_logger.debug('Forcing re-authentication with the provided credential')
force = True
# 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 = {
**self._base_headers,
'Authorization': f'Token {list(self._token.values())[0]}',
}
self.user = username
return
# Get a credential: a password, or failing that a token
if password is None and token is None:
password = getattr(self._par, 'ALYX_PWD', None)
if password is None and token is None:
if self.silent:
warnings.warn(
'No password, token or cached token in silent mode. '
'Please run the following to re-authenticate:\n\t'
'AlyxClient(silent=False).authenticate'
'(username=<username>, force=True)\n'
'An account without a password, such as one that signs in through an '
'identity provider, should pass a token instead:\n\t'
'AlyxClient(silent=True).authenticate'
'(username=<username>, token=<token>)',
UserWarning,
)
else:
# Offered as a fallback on the existing prompt rather than as a question of its
# own, so that the common case of having a password is unchanged.
password = getpass(
f'Enter Alyx password for "{username}" '
'(leave blank to enter an API token instead):')
if not password:
password = None
token = getpass(
f'Enter Alyx API token for "{username}", '
f'found at {self.base_url}/me:') or None
# Remove previous token
self._clear_token(username)
if token is not None:
self._token = token if isinstance(token, dict) else {'token': token}
else:
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 = {
**self._base_headers,
'Authorization': 'Token {}'.format(list(self._token.values())[0]),
}
if token is not None:
# A password gets a definitive yes or no from /auth-token; a token has no such
# endpoint, so check it before caching rather than letting a mistyped one surface
# as a puzzling failure on the first real query. The check also reports who the
# token belongs to, which is the one credential that can name its own owner.
username = self._resolve_token_user(username)
if cache_token:
# Update saved pars
par = one.params.get(client=self.base_url, silent=True)
tokens = getattr(par, 'TOKEN', {})
tokens[username] = self._token
par = par.set('TOKEN', tokens)
if not getattr(par, 'ALYX_LOGIN', None):
# The cached token is looked up by username, so record one that was resolved
# or prompted for, or it could never be found again.
par = par.set('ALYX_LOGIN', username)
one.params.save(par, self.base_url)
# Only fill in the login where this client has none: one built for a particular
# user keeps that user.
self._par = self._par.set('TOKEN', tokens)
if not getattr(self._par, 'ALYX_LOGIN', None):
self._par = self._par.set('ALYX_LOGIN', username)
self.user = username
if not self.silent:
print(f'Connected to {self.base_url} as user "{self.user}"')
def _verify_token(self, username=None):
"""Check that a user-supplied token is accepted, and discard it if not.
Only an explicit rejection counts as failure. Anything else - a network blip, an
endpoint missing on an older Alyx - leaves the token in place, because refusing a
credential that is probably fine is worse than letting the first real query report the
problem.
Parameters
----------
username : str, optional
The user the token was given for; its cached token is removed if the token is bad.
Returns
-------
str, None
The user the database says the token belongs to, where it says. None where the
question could not be answered, which is not in itself a failure.
Raises
------
requests.HTTPError
If the database explicitly rejects the token.
"""
try:
rep = requests.get(self.base_url + '/me', headers=self._headers,
timeout=self.token_verify_timeout)
except requests.exceptions.RequestException as ex:
_logger.debug('Could not verify the token (%s); assuming it is valid', ex)
return None
if rep.status_code in (401, 403):
self._clear_token(username)
whose = f' for user "{username}"' if username else ''
raise requests.HTTPError(
rep.status_code, rep.url,
f'Alyx rejected the API token provided{whose}. '
f'Tokens are shown at {self.base_url}/me and change when regenerated.',
response=rep)
_logger.debug('Token accepted for user "%s"', username)
try:
# Older databases serve /me as a web page, so only a JSON body names the user.
reported = rep.json().get('username')
except ValueError:
return None
# Only a non-empty string counts: the name becomes a key in the parameter file, and one
# that will not serialise truncates that file as it is written.
return reported if isinstance(reported, str) and reported else None
def _resolve_token_user(self, username):
"""Validate a token and settle on the username to cache it under.
A name the database reports wins over one the caller typed: caching a token under the
wrong name would leave the parameter file claiming an account it does not own.
Parameters
----------
username : str, None
The username the caller supplied, if any.
Returns
-------
str
The username to record.
Raises
------
requests.HTTPError
If the database explicitly rejects the token.
ValueError
If no username was given and the database would not name one.
"""
verified = self._verify_token(username)
if verified and username and verified != username:
warnings.warn(
f'The token provided belongs to "{verified}", not "{username}"; '
f'authenticating as "{verified}".', UserWarning)
username = verified or username
if not username:
self._clear_token(username)
raise ValueError(
'Could not determine the user this token belongs to. Pass a username, or '
f'check that {self.base_url} is an Alyx database recent enough to serve /me.')
return username
def _clear_token(self, username):
"""Remove auth token from client params.
Deletes the cached authentication token for a given user.
"""
par = one.params.get(client=self.base_url, silent=True)
# 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._token = None
if self._headers and 'Authorization' in self._headers:
del self._headers['Authorization']
[docs]
def logout(self):
"""Log out from Alyx.
Deletes the cached authentication token for the currently logged-in user
and clears the REST cache.
"""
if not self.is_logged_in:
return
self._clear_token(username := self.user)
self.user = None
self.clear_rest_cache()
if not self.silent:
print(f'{username} logged out from {self.base_url}')
[docs]
def delete(self, rest_query):
"""Send a DELETE request to the Alyx server.
Will raise an exception on any HTTP 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):
"""Download file(s) from data server from a REST file record URL.
Parameters
----------
url : str, list
Full url(s) of the file(s).
**kwargs
WebClient.http_download_file parameters.
Returns
-------
pathlib.Path, list of pathlib.Path
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):
"""Download 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.
"""
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 = None
if source.startswith(self.base_url):
if not self.is_logged_in:
self.authenticate()
headers = self._headers
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):
"""Assert 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):
"""Send a GET request to the Alyx server.
Will raise an exception on any HTTP 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 {'count', 'next', 'previous', 'results'} <= rep.keys():
# If a detail message is returned, log it as a warning
# This is not an error, but may be useful to the user
if detail := rep.pop('detail', None):
_logger.warning('Alyx server returned a detail message: %s', detail)
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):
"""Send 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):
"""Send 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):
"""Send 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 REST API wrapper.
If no arguments are passed, lists available endpoints.
Parameters
----------
url : str
Endpoint name.
action : str
One of 'list', 'create', 'read', 'update', 'partial_update', 'delete'.
id : str, uuid.UUID
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
c.f. https://openalyx.internationalbrainlab.org/docs/
Returns
-------
list, dict
List of queried dicts ('list') or dict (other actions).
Examples
--------
List available endpoint
>>> client = AlyxClient()
... client.rest()
List available actions for the 'subjects' endpoint
>>> client.rest('subjects')
Example REST endpoint with all actions
>>> 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')})
"""
# if endpoint is None, list available endpoints
if not url:
pprint(self.rest_schemes.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.endpoints:
av = [k for k in self.rest_schemes.endpoints if not k.startswith('_') and k]
raise ValueError(
'REST endpoint "'
+ endpoint
+ '" does not exist. Available '
+ 'endpoints are \n '
+ '\n '.join(av)
)
# 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(self.rest_schemes.actions(endpoint))
self.rest_schemes.print_endpoint_info(endpoint)
return
if action == 'partial-update':
# Endpoint names are hyphenated but action names are underscored;
# convert for user convenience
action = 'partial_update'
# make sure the desired action exists, if not throw an informative error
if action not in self.rest_schemes.actions(endpoint):
raise ValueError(
'Action "'
+ action
+ '" for REST endpoint "'
+ endpoint
+ '" does '
+ 'not exist. Available actions are: '
+ '\n '
+ '\n '.join(self.rest_schemes.actions(endpoint))
)
# 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: '
+ self.rest_schemes.url(endpoint, action)
)
return
# the actions below require a data dictionary, warn and help the user with fields list
if action in ['create', 'update', 'partial_update'] and not data:
rest_params = self.rest_schemes.fields(endpoint, action)
pprint(rest_params)
_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':
if id:
# this is a special case of the list where we query a uuid
# usually read is better but list may return fewer data and therefore be faster
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:
# if django arg is present but is None, server will return a cryptic 500 status
if 'django' in kwargs and kwargs['django'] is None:
del kwargs['django']
# Convert all lists in query params to comma separated list
if len(set(kwargs.keys()) - set(self.rest_schemes.field_names(endpoint, action))):
missing = set(kwargs.keys()) - set(
self.rest_schemes.field_names(endpoint, action))
warnings.warn(
f"Error: Unsupported fields '{missing}' in query parameters"
f" for endpoint {endpoint}.")
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':
return self.get('/' + endpoint + '/' + id.split('/')[-1], **cache_args)
elif action == 'create':
return self.post('/' + endpoint, data=data, files=files)
elif action == 'delete':
return self.delete('/' + endpoint + '/' + id.split('/')[-1])
elif action == 'partial_update':
return self.patch('/' + endpoint + '/' + id.split('/')[-1], data=data, files=files)
elif action == 'update':
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.endpoints:
av = (k for k in self.rest_schemes.endpoints 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.
NOTE: Destructive write! WILL NOT CHECK IF DATA EXISTS
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.rest_cache_dir.glob('*'):
file.unlink()