intern.service.boss.v1.project module
# Copyright 2016 The Johns Hopkins University Applied Physics Laboratory
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from intern.service.boss import BaseVersion
from intern.service.boss.v1 import BOSS_API_VERSION
from intern.resource.boss.resource import *
from requests import HTTPError
import copy
class ProjectService_1(BaseVersion):
"""The Boss API v1 project service.
"""
def __init__(self):
BaseVersion.__init__(self)
@property
def version(self):
"""Return the API Version for this implementation
"""
return BOSS_API_VERSION
def list_groups(self, filtr, url_prefix, auth, session, send_opts):
"""Get the groups the logged in user is a member of.
Optionally filter by 'member' or 'maintainer'.
Args:
filtr (string|None): ['member'|'maintainer'] or defaults to None.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[string]): List of group names.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'GET', 'application/json', url_prefix, auth)
if filtr is not None:
if not filtr == 'member' and not filtr == 'maintainer':
raise RuntimeError(
'filtr must be either "member", "maintainer", or None.')
req.params = {'filter': filtr}
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['groups']
msg = ('List groups failed, got HTTP response: ({}) - {}'.format(
resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_group(self, name, user_name, url_prefix, auth, session, send_opts):
"""Get owner of group and the resources it's attached to.
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(dict): Keys include 'owner', 'name', 'resources'.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'GET', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = ('Get failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def create_group(self, name, url_prefix, auth, session, send_opts):
"""Create a new group.
Args:
name (string): Name of the group to create.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'POST', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = ('Create failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group(self, name, url_prefix, auth, session, send_opts):
"""Delete given group.
Args:
name (string): Name of group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'DELETE', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Delete failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list_group_members(self, name, url_prefix, auth, session, send_opts):
"""Get the members of a group (does not include maintainers).
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[string]): List of member names.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'GET', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['members']
msg = ('Failed getting members of group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_is_group_member(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Check if the given user is a member of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(bool): False if user not a member.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'GET', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['result']
msg = ('Failed determining if user {} is member of group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_group_member(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'POST', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed adding user {} to group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group_member(
self, grp_name, user, url_prefix, auth, session, send_opts):
"""Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'DELETE', 'application/json', url_prefix, auth,
grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting maintainer {} from group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list_group_maintainers(self, name, url_prefix, auth, session, send_opts):
"""Get the maintainers of a group.
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[string]): List of maintainer names.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'GET', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['maintainers']
msg = ('Failed getting maintainers of group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_is_group_maintainer(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Check if the given user is a maintainer of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(bool): False if user not a maintainer.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'GET', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['result']
msg = ('Failed determining if user {} is maintainer of group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_group_maintainer(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'POST', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed adding maintainer {} to group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group_maintainer(
self, grp_name, user, url_prefix, auth, session, send_opts):
"""Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'DELETE', 'application/json', url_prefix, auth,
grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting maintainer {} from group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list_permissions(self, group_name=None, resource=None,
url_prefix=None, auth=None, session=None, send_opts=None):
"""List the permission sets for the logged in user
Optionally filter by resource or group.
Args:
group_name (string): Name of group to filter on
resource (intern.resource.boss.BossResource): Identifies which data model object to filter on
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[dict]): List of dictionaries of permission sets
"""
filter_params = {}
if group_name:
filter_params["group"] = group_name
if resource:
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('GET', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 200:
msg = "Failed to get permission sets. "
if group_name:
msg = "{} Group: {}".format(msg, group_name)
if resource:
msg = "{} Resource: {}".format(msg, resource.name)
msg = '{}, got HTTP response: ({}) - {}'.format(msg, resp.status_code, resp.text)
raise HTTPError(msg, request=req, response=resp)
else:
return resp.json()["permission-sets"]
def get_permissions(self, group_name, resource,url_prefix, auth, session, send_opts):
"""Get the permission set for a specific group/resource combination
Args:
group_name (string): Name of group to filter on
resource (intern.resource.boss.BossResource): Identifies which data model object to filter on
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[str]): List of permissions
"""
filter_params = {"group": group_name}
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('GET', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 200:
msg = "Failed to get permission set for Group: {} Resource: {}".format(group_name, resource.name)
msg = '{}, got HTTP response: ({}) - {}'.format(msg, resp.status_code, resp.text)
raise HTTPError(msg, request=req, response=resp)
else:
if resp.json()["permission-sets"]:
return resp.json()["permission-sets"][0]['permissions']
else:
return []
def add_permissions(self, group_name, resource, permissions, url_prefix, auth, session, send_opts):
"""
Args:
group_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
permissions (list): List of permissions to add to the given resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
"""
post_data = {"group": group_name,
"permissions": permissions,
}
post_data.update(resource.get_dict_route())
req = self.get_permission_request('POST', 'application/json',
url_prefix, auth, post_data=post_data)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 201:
msg = ('Failed adding permissions to group {}, got HTTP response: ({}) - {}'.format(group_name,
resp.status_code,
resp.text))
raise HTTPError(msg, request=req, response=resp)
def update_permissions(self, group_name, resource, permissions, url_prefix, auth, session, send_opts):
"""
Args:
group_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
permissions (list): List of permissions to attach to the given resource. Will overwrite existing permissions
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
"""
post_data = {"group": group_name,
"permissions": permissions,
}
post_data.update(resource.get_dict_route())
req = self.get_permission_request('PATCH', 'application/json',
url_prefix, auth, post_data=post_data)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 200:
msg = ('Failed adding permissions to group {}, got HTTP response: ({}) - {}'.format(group_name,
resp.status_code,
resp.text))
raise HTTPError(msg, request=req, response=resp)
def delete_permissions(self, grp_name, resource, url_prefix, auth, session, send_opts):
"""
Args:
grp_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
filter_params = {"group": grp_name}
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('DELETE', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting permissions to group {}, got HTTP response: ({}) - {}'.format(
grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request=req, response=resp)
def get_user_roles(self, user, url_prefix, auth, session, send_opts):
"""Get roles associated with the given user.
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list): List of roles that user has.
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'GET', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = (
'Failed getting roles for user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_user_role(self, user, role, url_prefix, auth, session, send_opts):
"""Add role to given user.
Args:
user (string): User name.
role (string): Role to assign.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'POST', 'application/json', url_prefix, auth,
user, role)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = (
'Failed adding role: {} to user: {}, got HTTP response: ({}) - {}'
.format(role, user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_user_role(self, user, role, url_prefix, auth, session, send_opts):
"""Remove role from given user.
Args:
user (string): User name.
role (string): Role to remove.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'DELETE', 'application/json', url_prefix, auth,
user, role)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = (
'Failed deleting role: {} from user: {}, got HTTP response: ({}) - {}'
.format(role, user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_user(self, user, url_prefix, auth, session, send_opts):
"""Get user's data (first and last name, email, etc).
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(dictionary): User's data encoded in a dictionary.
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'GET', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = (
'Failed getting user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request=req, response=resp)
def add_user(
self, user, first_name, last_name, email, password,
url_prefix, auth, session, send_opts):
"""Add a new user.
Args:
user (string): User name.
first_name (string): User's first name.
last_name (string): User's last name.
email: (string): User's email address.
password: (string): User's password.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'POST', 'application/json', url_prefix, auth,
user, first_name, last_name, email, password)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = (
'Failed adding user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_user(self, user, url_prefix, auth, session, send_opts):
"""Delete the given user.
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'DELETE', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = (
'Failed deleting user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list(self, resource, url_prefix, auth, session, send_opts):
"""List all resources of the same type as the given resource.
Args:
resource (intern.resource.boss.BossResource): List resources of the same type as this..
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list): List of resources. Each resource is a dictionary.
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'GET', 'application/json', url_prefix, auth,
proj_list_req=True)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return self._get_resource_list(resp.json())
err = ('List failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def create(self, resource, url_prefix, auth, session, send_opts):
"""Create the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises:
requests.HTTPError on failure.
"""
json = self._get_resource_params(resource)
req = self.get_request(resource, 'POST', 'application/json', url_prefix, auth, json=json)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return self._create_resource_from_dict(resource, resp.json())
err = ('Create failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def get(self, resource, url_prefix, auth, session, send_opts):
"""Get attributes of the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'GET', 'application/json', url_prefix, auth)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return self._create_resource_from_dict(resource, resp.json())
err = ('Get failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def update(self, resource_name, resource, url_prefix, auth, session, send_opts):
"""Updates an entity in the data model using the given resource.
Args:
resource_name (string): Current name of the resource (in case the resource is getting its name changed).
resource (intern.resource.boss.BossResource): New attributes for the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns updated resource of given type on success.
Raises:
requests.HTTPError on failure.
"""
# Create a copy of the resource and change its name to resource_name
# in case the update includes changing the name of a resource.
old_resource = copy.deepcopy(resource)
old_resource.name = resource_name
json = self._get_resource_params(resource, for_update=True)
req = self.get_request(old_resource, 'PUT', 'application/json', url_prefix, auth, json=json)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return self._create_resource_from_dict(resource, resp.json())
err = ('Update failed on {}, got HTTP response: ({}) - {}'.format(
old_resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def delete(self, resource, url_prefix, auth, session, send_opts):
"""Deletes the entity described by the given resource.
Args:
resource (intern.resource.boss.BossResource)
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'DELETE', 'application/json', url_prefix, auth)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
err = ('Delete failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def _get_resource_params(self, resource, for_update=False):
"""Get dictionary containing all parameters for the given resource.
When getting params for a coordinate frame update, only name and
description are returned because they are the only fields that can
be updated.
Args:
resource (intern.resource.boss.resource.BossResource): A sub-class
whose parameters will be extracted into a dictionary.
for_update (bool): True if params will be used for an update.
Returns:
(dictionary): A dictionary containing the resource's parameters as
required by the Boss API.
Raises:
TypeError if resource is not a supported class.
"""
if isinstance(resource, CollectionResource):
return self._get_collection_params(resource)
if isinstance(resource, ExperimentResource):
return self._get_experiment_params(resource, for_update)
if isinstance(resource, CoordinateFrameResource):
return self._get_coordinate_params(resource, for_update)
if isinstance(resource, ChannelResource):
return self._get_channel_params(resource, for_update)
raise TypeError('resource is not supported type.')
def _get_collection_params(self, coll):
return { 'name': coll.name, 'description': coll.description }
def _get_experiment_params(self, exp, for_update):
if not for_update:
return {
'name': exp.name,
'description': exp.description ,
'coord_frame': exp.coord_frame,
'num_hierarchy_levels': exp.num_hierarchy_levels,
'hierarchy_method': exp.hierarchy_method,
'num_time_samples': exp.num_time_samples,
'collection': exp.coll_name,
'time_step': exp.time_step,
'time_step_unit': exp.time_step_unit
}
return {
'name': exp.name,
'description': exp.description ,
'num_hierarchy_levels': exp.num_hierarchy_levels,
'hierarchy_method': exp.hierarchy_method
}
def _get_coordinate_params(self, coord, for_update):
if not for_update:
return {
'name': coord.name,
'description': coord.description ,
'x_start': coord.x_start,
'x_stop': coord.x_stop,
'y_start': coord.y_start,
'y_stop': coord.y_stop,
'z_start': coord.z_start,
'z_stop': coord.z_stop,
'x_voxel_size': coord.x_voxel_size,
'y_voxel_size': coord.y_voxel_size,
'z_voxel_size': coord.z_voxel_size,
'voxel_unit': coord.voxel_unit
}
return { 'name': coord.name, 'description': coord.description }
def _get_channel_params(self, chan, for_update):
if not for_update:
return {
'name': chan.name,
'description': chan.description ,
'default_time_sample': chan.default_time_sample,
'datatype': chan.datatype,
'base_resolution': chan.base_resolution,
'type': chan.type,
'sources': chan.sources,
'related': chan.related
}
return {
'name': chan.name,
'description': chan.description,
'base_resolution': chan.base_resolution,
'sources': chan.sources,
'related': chan.related
}
def _create_resource_from_dict(self, resource, dict):
"""
Args:
resource (intern.resource.boss.BossResource): Used to determine type of resource to create.
dict (dictionary): JSON data returned by the Boss API.
Returns:
(intern.resource.boss.BossResource): Instance populated with values from dict.
Raises:
KeyError if dict missing required key.
TypeError if resource is not a supported class.
"""
if isinstance(resource, CollectionResource):
return self._get_collection(dict)
if isinstance(resource, ExperimentResource):
return self._get_experiment(dict, resource.coll_name)
if isinstance(resource, CoordinateFrameResource):
return self._get_coordinate(dict)
if isinstance(resource, ChannelResource):
return self._get_channel(dict, resource.coll_name, resource.exp_name)
raise TypeError('resource is not supported type.')
def _get_collection(self, dict):
name = dict['name']
description = dict['description']
creator = dict['creator']
return CollectionResource(
name, description, creator, raw=dict)
def _get_experiment(self, dict, coll_name):
exp_keys = [
'name', 'description', 'creator', 'coord_frame',
'num_hierarchy_levels', 'hierarchy_method', 'num_time_samples',
'time_step', 'time_step_unit'
]
filtered = { k:v for (k, v) in dict.items() if k in exp_keys }
return ExperimentResource(
collection_name=coll_name, raw=dict, **filtered)
def _get_coordinate(self, dict):
coord_keys = [
'name', 'description', 'x_start', 'x_stop',
'y_start', 'y_stop', 'z_start', 'z_stop',
'x_voxel_size', 'y_voxel_size', 'z_voxel_size',
'voxel_unit'
]
filtered = { k:v for (k, v) in dict.items() if k in coord_keys }
return CoordinateFrameResource(raw=dict, **filtered)
def _get_channel(self, dict, coll_name, exp_name):
chan_keys = [
'name', 'description', 'creator', 'default_time_sample',
'datatype', 'base_resolution', 'type', 'sources', 'related'
]
filtered = { k:v for (k, v) in dict.items() if k in chan_keys }
collection = coll_name
return ChannelResource(
collection_name=collection,
experiment_name=exp_name, raw=dict, **filtered)
def _get_resource_list(self, rsrc_dict):
"""Extracts list of resources from the HTTP response.
Args:
rsrc_dict (dict): HTTP response encoded in a dictionary.
Returns:
(list[string]): List of a type of resource (collections, experiments, etc).
Raises:
(RuntimeError): If rsrc_dict does not contain any known resources.
"""
if 'collections' in rsrc_dict:
return rsrc_dict['collections']
if 'experiments' in rsrc_dict:
return rsrc_dict['experiments']
if 'channels' in rsrc_dict:
return rsrc_dict['channels']
if 'coords' in rsrc_dict:
return rsrc_dict['coords']
raise RuntimeError('Invalid list response received from Boss. No known resource type returned.')
Module variables
var BOSS_API_VERSION
Classes
class ProjectService_1
The Boss API v1 project service.
class ProjectService_1(BaseVersion):
"""The Boss API v1 project service.
"""
def __init__(self):
BaseVersion.__init__(self)
@property
def version(self):
"""Return the API Version for this implementation
"""
return BOSS_API_VERSION
def list_groups(self, filtr, url_prefix, auth, session, send_opts):
"""Get the groups the logged in user is a member of.
Optionally filter by 'member' or 'maintainer'.
Args:
filtr (string|None): ['member'|'maintainer'] or defaults to None.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[string]): List of group names.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'GET', 'application/json', url_prefix, auth)
if filtr is not None:
if not filtr == 'member' and not filtr == 'maintainer':
raise RuntimeError(
'filtr must be either "member", "maintainer", or None.')
req.params = {'filter': filtr}
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['groups']
msg = ('List groups failed, got HTTP response: ({}) - {}'.format(
resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_group(self, name, user_name, url_prefix, auth, session, send_opts):
"""Get owner of group and the resources it's attached to.
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(dict): Keys include 'owner', 'name', 'resources'.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'GET', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = ('Get failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def create_group(self, name, url_prefix, auth, session, send_opts):
"""Create a new group.
Args:
name (string): Name of the group to create.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'POST', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = ('Create failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group(self, name, url_prefix, auth, session, send_opts):
"""Delete given group.
Args:
name (string): Name of group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'DELETE', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Delete failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list_group_members(self, name, url_prefix, auth, session, send_opts):
"""Get the members of a group (does not include maintainers).
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[string]): List of member names.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'GET', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['members']
msg = ('Failed getting members of group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_is_group_member(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Check if the given user is a member of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(bool): False if user not a member.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'GET', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['result']
msg = ('Failed determining if user {} is member of group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_group_member(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'POST', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed adding user {} to group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group_member(
self, grp_name, user, url_prefix, auth, session, send_opts):
"""Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'DELETE', 'application/json', url_prefix, auth,
grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting maintainer {} from group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list_group_maintainers(self, name, url_prefix, auth, session, send_opts):
"""Get the maintainers of a group.
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[string]): List of maintainer names.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'GET', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['maintainers']
msg = ('Failed getting maintainers of group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_is_group_maintainer(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Check if the given user is a maintainer of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(bool): False if user not a maintainer.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'GET', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['result']
msg = ('Failed determining if user {} is maintainer of group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_group_maintainer(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'POST', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed adding maintainer {} to group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group_maintainer(
self, grp_name, user, url_prefix, auth, session, send_opts):
"""Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'DELETE', 'application/json', url_prefix, auth,
grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting maintainer {} from group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list_permissions(self, group_name=None, resource=None,
url_prefix=None, auth=None, session=None, send_opts=None):
"""List the permission sets for the logged in user
Optionally filter by resource or group.
Args:
group_name (string): Name of group to filter on
resource (intern.resource.boss.BossResource): Identifies which data model object to filter on
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[dict]): List of dictionaries of permission sets
"""
filter_params = {}
if group_name:
filter_params["group"] = group_name
if resource:
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('GET', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 200:
msg = "Failed to get permission sets. "
if group_name:
msg = "{} Group: {}".format(msg, group_name)
if resource:
msg = "{} Resource: {}".format(msg, resource.name)
msg = '{}, got HTTP response: ({}) - {}'.format(msg, resp.status_code, resp.text)
raise HTTPError(msg, request=req, response=resp)
else:
return resp.json()["permission-sets"]
def get_permissions(self, group_name, resource,url_prefix, auth, session, send_opts):
"""Get the permission set for a specific group/resource combination
Args:
group_name (string): Name of group to filter on
resource (intern.resource.boss.BossResource): Identifies which data model object to filter on
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[str]): List of permissions
"""
filter_params = {"group": group_name}
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('GET', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 200:
msg = "Failed to get permission set for Group: {} Resource: {}".format(group_name, resource.name)
msg = '{}, got HTTP response: ({}) - {}'.format(msg, resp.status_code, resp.text)
raise HTTPError(msg, request=req, response=resp)
else:
if resp.json()["permission-sets"]:
return resp.json()["permission-sets"][0]['permissions']
else:
return []
def add_permissions(self, group_name, resource, permissions, url_prefix, auth, session, send_opts):
"""
Args:
group_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
permissions (list): List of permissions to add to the given resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
"""
post_data = {"group": group_name,
"permissions": permissions,
}
post_data.update(resource.get_dict_route())
req = self.get_permission_request('POST', 'application/json',
url_prefix, auth, post_data=post_data)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 201:
msg = ('Failed adding permissions to group {}, got HTTP response: ({}) - {}'.format(group_name,
resp.status_code,
resp.text))
raise HTTPError(msg, request=req, response=resp)
def update_permissions(self, group_name, resource, permissions, url_prefix, auth, session, send_opts):
"""
Args:
group_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
permissions (list): List of permissions to attach to the given resource. Will overwrite existing permissions
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
"""
post_data = {"group": group_name,
"permissions": permissions,
}
post_data.update(resource.get_dict_route())
req = self.get_permission_request('PATCH', 'application/json',
url_prefix, auth, post_data=post_data)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 200:
msg = ('Failed adding permissions to group {}, got HTTP response: ({}) - {}'.format(group_name,
resp.status_code,
resp.text))
raise HTTPError(msg, request=req, response=resp)
def delete_permissions(self, grp_name, resource, url_prefix, auth, session, send_opts):
"""
Args:
grp_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
filter_params = {"group": grp_name}
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('DELETE', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting permissions to group {}, got HTTP response: ({}) - {}'.format(
grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request=req, response=resp)
def get_user_roles(self, user, url_prefix, auth, session, send_opts):
"""Get roles associated with the given user.
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list): List of roles that user has.
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'GET', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = (
'Failed getting roles for user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_user_role(self, user, role, url_prefix, auth, session, send_opts):
"""Add role to given user.
Args:
user (string): User name.
role (string): Role to assign.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'POST', 'application/json', url_prefix, auth,
user, role)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = (
'Failed adding role: {} to user: {}, got HTTP response: ({}) - {}'
.format(role, user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_user_role(self, user, role, url_prefix, auth, session, send_opts):
"""Remove role from given user.
Args:
user (string): User name.
role (string): Role to remove.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'DELETE', 'application/json', url_prefix, auth,
user, role)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = (
'Failed deleting role: {} from user: {}, got HTTP response: ({}) - {}'
.format(role, user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_user(self, user, url_prefix, auth, session, send_opts):
"""Get user's data (first and last name, email, etc).
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(dictionary): User's data encoded in a dictionary.
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'GET', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = (
'Failed getting user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request=req, response=resp)
def add_user(
self, user, first_name, last_name, email, password,
url_prefix, auth, session, send_opts):
"""Add a new user.
Args:
user (string): User name.
first_name (string): User's first name.
last_name (string): User's last name.
email: (string): User's email address.
password: (string): User's password.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'POST', 'application/json', url_prefix, auth,
user, first_name, last_name, email, password)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = (
'Failed adding user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_user(self, user, url_prefix, auth, session, send_opts):
"""Delete the given user.
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'DELETE', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = (
'Failed deleting user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def list(self, resource, url_prefix, auth, session, send_opts):
"""List all resources of the same type as the given resource.
Args:
resource (intern.resource.boss.BossResource): List resources of the same type as this..
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list): List of resources. Each resource is a dictionary.
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'GET', 'application/json', url_prefix, auth,
proj_list_req=True)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return self._get_resource_list(resp.json())
err = ('List failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def create(self, resource, url_prefix, auth, session, send_opts):
"""Create the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises:
requests.HTTPError on failure.
"""
json = self._get_resource_params(resource)
req = self.get_request(resource, 'POST', 'application/json', url_prefix, auth, json=json)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return self._create_resource_from_dict(resource, resp.json())
err = ('Create failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def get(self, resource, url_prefix, auth, session, send_opts):
"""Get attributes of the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'GET', 'application/json', url_prefix, auth)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return self._create_resource_from_dict(resource, resp.json())
err = ('Get failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def update(self, resource_name, resource, url_prefix, auth, session, send_opts):
"""Updates an entity in the data model using the given resource.
Args:
resource_name (string): Current name of the resource (in case the resource is getting its name changed).
resource (intern.resource.boss.BossResource): New attributes for the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns updated resource of given type on success.
Raises:
requests.HTTPError on failure.
"""
# Create a copy of the resource and change its name to resource_name
# in case the update includes changing the name of a resource.
old_resource = copy.deepcopy(resource)
old_resource.name = resource_name
json = self._get_resource_params(resource, for_update=True)
req = self.get_request(old_resource, 'PUT', 'application/json', url_prefix, auth, json=json)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return self._create_resource_from_dict(resource, resp.json())
err = ('Update failed on {}, got HTTP response: ({}) - {}'.format(
old_resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def delete(self, resource, url_prefix, auth, session, send_opts):
"""Deletes the entity described by the given resource.
Args:
resource (intern.resource.boss.BossResource)
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'DELETE', 'application/json', url_prefix, auth)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
err = ('Delete failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def _get_resource_params(self, resource, for_update=False):
"""Get dictionary containing all parameters for the given resource.
When getting params for a coordinate frame update, only name and
description are returned because they are the only fields that can
be updated.
Args:
resource (intern.resource.boss.resource.BossResource): A sub-class
whose parameters will be extracted into a dictionary.
for_update (bool): True if params will be used for an update.
Returns:
(dictionary): A dictionary containing the resource's parameters as
required by the Boss API.
Raises:
TypeError if resource is not a supported class.
"""
if isinstance(resource, CollectionResource):
return self._get_collection_params(resource)
if isinstance(resource, ExperimentResource):
return self._get_experiment_params(resource, for_update)
if isinstance(resource, CoordinateFrameResource):
return self._get_coordinate_params(resource, for_update)
if isinstance(resource, ChannelResource):
return self._get_channel_params(resource, for_update)
raise TypeError('resource is not supported type.')
def _get_collection_params(self, coll):
return { 'name': coll.name, 'description': coll.description }
def _get_experiment_params(self, exp, for_update):
if not for_update:
return {
'name': exp.name,
'description': exp.description ,
'coord_frame': exp.coord_frame,
'num_hierarchy_levels': exp.num_hierarchy_levels,
'hierarchy_method': exp.hierarchy_method,
'num_time_samples': exp.num_time_samples,
'collection': exp.coll_name,
'time_step': exp.time_step,
'time_step_unit': exp.time_step_unit
}
return {
'name': exp.name,
'description': exp.description ,
'num_hierarchy_levels': exp.num_hierarchy_levels,
'hierarchy_method': exp.hierarchy_method
}
def _get_coordinate_params(self, coord, for_update):
if not for_update:
return {
'name': coord.name,
'description': coord.description ,
'x_start': coord.x_start,
'x_stop': coord.x_stop,
'y_start': coord.y_start,
'y_stop': coord.y_stop,
'z_start': coord.z_start,
'z_stop': coord.z_stop,
'x_voxel_size': coord.x_voxel_size,
'y_voxel_size': coord.y_voxel_size,
'z_voxel_size': coord.z_voxel_size,
'voxel_unit': coord.voxel_unit
}
return { 'name': coord.name, 'description': coord.description }
def _get_channel_params(self, chan, for_update):
if not for_update:
return {
'name': chan.name,
'description': chan.description ,
'default_time_sample': chan.default_time_sample,
'datatype': chan.datatype,
'base_resolution': chan.base_resolution,
'type': chan.type,
'sources': chan.sources,
'related': chan.related
}
return {
'name': chan.name,
'description': chan.description,
'base_resolution': chan.base_resolution,
'sources': chan.sources,
'related': chan.related
}
def _create_resource_from_dict(self, resource, dict):
"""
Args:
resource (intern.resource.boss.BossResource): Used to determine type of resource to create.
dict (dictionary): JSON data returned by the Boss API.
Returns:
(intern.resource.boss.BossResource): Instance populated with values from dict.
Raises:
KeyError if dict missing required key.
TypeError if resource is not a supported class.
"""
if isinstance(resource, CollectionResource):
return self._get_collection(dict)
if isinstance(resource, ExperimentResource):
return self._get_experiment(dict, resource.coll_name)
if isinstance(resource, CoordinateFrameResource):
return self._get_coordinate(dict)
if isinstance(resource, ChannelResource):
return self._get_channel(dict, resource.coll_name, resource.exp_name)
raise TypeError('resource is not supported type.')
def _get_collection(self, dict):
name = dict['name']
description = dict['description']
creator = dict['creator']
return CollectionResource(
name, description, creator, raw=dict)
def _get_experiment(self, dict, coll_name):
exp_keys = [
'name', 'description', 'creator', 'coord_frame',
'num_hierarchy_levels', 'hierarchy_method', 'num_time_samples',
'time_step', 'time_step_unit'
]
filtered = { k:v for (k, v) in dict.items() if k in exp_keys }
return ExperimentResource(
collection_name=coll_name, raw=dict, **filtered)
def _get_coordinate(self, dict):
coord_keys = [
'name', 'description', 'x_start', 'x_stop',
'y_start', 'y_stop', 'z_start', 'z_stop',
'x_voxel_size', 'y_voxel_size', 'z_voxel_size',
'voxel_unit'
]
filtered = { k:v for (k, v) in dict.items() if k in coord_keys }
return CoordinateFrameResource(raw=dict, **filtered)
def _get_channel(self, dict, coll_name, exp_name):
chan_keys = [
'name', 'description', 'creator', 'default_time_sample',
'datatype', 'base_resolution', 'type', 'sources', 'related'
]
filtered = { k:v for (k, v) in dict.items() if k in chan_keys }
collection = coll_name
return ChannelResource(
collection_name=collection,
experiment_name=exp_name, raw=dict, **filtered)
def _get_resource_list(self, rsrc_dict):
"""Extracts list of resources from the HTTP response.
Args:
rsrc_dict (dict): HTTP response encoded in a dictionary.
Returns:
(list[string]): List of a type of resource (collections, experiments, etc).
Raises:
(RuntimeError): If rsrc_dict does not contain any known resources.
"""
if 'collections' in rsrc_dict:
return rsrc_dict['collections']
if 'experiments' in rsrc_dict:
return rsrc_dict['experiments']
if 'channels' in rsrc_dict:
return rsrc_dict['channels']
if 'coords' in rsrc_dict:
return rsrc_dict['coords']
raise RuntimeError('Invalid list response received from Boss. No known resource type returned.')
Ancestors (in MRO)
- ProjectService_1
- intern.service.boss.baseversion.BaseVersion
- __builtin__.object
Instance variables
var version
Return the API Version for this implementation
Methods
def __init__(
self)
def __init__(self):
BaseVersion.__init__(self)
def add_group_maintainer(
self, grp_name, user, url_prefix, auth, session, send_opts)
Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def add_group_maintainer(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'POST', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed adding maintainer {} to group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_group_member(
self, grp_name, user, url_prefix, auth, session, send_opts)
Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def add_group_member(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Add the given user to the named group.
Both group and user must already exist for this to succeed.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'POST', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed adding user {} to group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_permissions(
self, group_name, resource, permissions, url_prefix, auth, session, send_opts)
Args:
group_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
permissions (list): List of permissions to add to the given resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
def add_permissions(self, group_name, resource, permissions, url_prefix, auth, session, send_opts):
"""
Args:
group_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
permissions (list): List of permissions to add to the given resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
"""
post_data = {"group": group_name,
"permissions": permissions,
}
post_data.update(resource.get_dict_route())
req = self.get_permission_request('POST', 'application/json',
url_prefix, auth, post_data=post_data)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 201:
msg = ('Failed adding permissions to group {}, got HTTP response: ({}) - {}'.format(group_name,
resp.status_code,
resp.text))
raise HTTPError(msg, request=req, response=resp)
def add_user(
self, user, first_name, last_name, email, password, url_prefix, auth, session, send_opts)
Add a new user.
Args:
user (string): User name.
first_name (string): User's first name.
last_name (string): User's last name.
email: (string): User's email address.
password: (string): User's password.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def add_user(
self, user, first_name, last_name, email, password,
url_prefix, auth, session, send_opts):
"""Add a new user.
Args:
user (string): User name.
first_name (string): User's first name.
last_name (string): User's last name.
email: (string): User's email address.
password: (string): User's password.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'POST', 'application/json', url_prefix, auth,
user, first_name, last_name, email, password)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = (
'Failed adding user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def add_user_role(
self, user, role, url_prefix, auth, session, send_opts)
Add role to given user.
Args:
user (string): User name.
role (string): Role to assign.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def add_user_role(self, user, role, url_prefix, auth, session, send_opts):
"""Add role to given user.
Args:
user (string): User name.
role (string): Role to assign.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'POST', 'application/json', url_prefix, auth,
user, role)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = (
'Failed adding role: {} to user: {}, got HTTP response: ({}) - {}'
.format(role, user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def build_cutout_url(
self, resource, url_prefix, resolution, x_range, y_range, z_range, time_range=None, id_list=[])
Build the url to access the cutout function of the Boss' volume service.
Args:
url_prefix (string): Do not end with a slash. Example of expected value: https://api.theboss.io
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (optional [list[int]]): time range such as [30, 40] which means t>=30 and t<40.
id_list (optional [list[int]]): list of object ids to filter the cutout by.
Returns
(string): Full URL to access API.
Raises
(RuntimeError): if *_range invalid.
def build_cutout_url(
self, resource, url_prefix, resolution, x_range, y_range, z_range, time_range=None, id_list=[]):
"""Build the url to access the cutout function of the Boss' volume service.
Args:
url_prefix (string): Do not end with a slash. Example of expected value: https://api.theboss.io
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (optional [list[int]]): time range such as [30, 40] which means t>=30 and t<40.
id_list (optional [list[int]]): list of object ids to filter the cutout by.
Returns:
(string): Full URL to access API.
Raises:
(RuntimeError): if *_range invalid.
"""
baseUrl = self.build_url(resource, url_prefix, 'cutout', req_type='cutout')
x_rng_lst = self.convert_int_list_range_to_str(x_range)
y_rng_lst = self.convert_int_list_range_to_str(y_range)
z_rng_lst = self.convert_int_list_range_to_str(z_range)
urlWithParams = (
baseUrl + '/' + str(resolution) + '/' + x_rng_lst + '/' + y_rng_lst +
'/' + z_rng_lst + '/')
if time_range is not None:
t_rng_lst = self.convert_int_list_range_to_str(time_range)
urlWithParams += t_rng_lst + '/'
if len(id_list) > 0:
urlWithParams += '?filter=' + self.convert_int_list_to_comma_sep_str(id_list)
return urlWithParams
def build_ids_url(
self, resource, url_prefix, resolution, x_range, y_range, z_range, time_range=None)
Build the url to access the ids function of the Boss' volume service.
 Example of expected value: https://api.theboss.io
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (optional [list[int]]): time range such as [30, 40] which means t>=30 and t<40.
Returns
(string): Full URL to access API.
Raises
(RuntimeError): if *_range invalid.
def build_ids_url(
self, resource, url_prefix, resolution,
x_range, y_range, z_range, time_range=None):
"""Build the url to access the ids function of the Boss' volume service.
 Example of expected value: https://api.theboss.io
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (optional [list[int]]): time range such as [30, 40] which means t>=30 and t<40.
Returns:
(string): Full URL to access API.
Raises:
(RuntimeError): if *_range invalid.
"""
base_url = self.build_url(resource, url_prefix, 'ids', req_type='cutout')
x_rng_lst = self.convert_int_list_range_to_str(x_range)
y_rng_lst = self.convert_int_list_range_to_str(y_range)
z_rng_lst = self.convert_int_list_range_to_str(z_range)
url_with_params = (
base_url + '/' + str(resolution) + '/' + x_rng_lst + '/' + y_rng_lst +
'/' + z_rng_lst + '/')
if time_range is not None:
t_rng_lst = self.convert_int_list_range_to_str(time_range)
url_with_params += t_rng_lst + '/'
return url_with_params
def build_metadata_url(
self, resource, url_prefix, key, value=None)
Build the url to access the Boss' metadata service.
Args:
url_prefix (string): Do not end with a slash. Example of expected value: https://api.theboss.io
key (string): Key to get/create/update/delete.
value (string): Value to assign to key or None.
Returns
(string): Full URL to access API.
def build_metadata_url(self, resource, url_prefix, key, value=None):
"""Build the url to access the Boss' metadata service.
Args:
url_prefix (string): Do not end with a slash. Example of expected value: https://api.theboss.io
key (string): Key to get/create/update/delete.
value (string): Value to assign to key or None.
Returns:
(string): Full URL to access API.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
suffix = resource.get_meta_route()
urlNoParams = (url_prefix + '/' + self.version + '/meta/' + suffix)
if key is None:
return urlNoParams
urlWithKey = urlNoParams + '/?key=' + key
if value is None:
return urlWithKey
return urlWithKey + '&value=' + str(value)
def build_url(
self, resource, url_prefix, service, req_type='normal')
Build the url to access the Boss' project service.
Args:
url_prefix (string): Do not end with a slash. Example of expected value: https://api.theboss.io
service (string): Name of service to access, such as collection, meta, coord, etc.
req_type (optional[string]): Valid values ['normal', 'list', 'cutout']. Defaults to 'normal'.
Returns
(string): Full URL to access API.
Raises
(RuntimeError): if invalid req_type given or url_prefix not given.
def build_url(self, resource, url_prefix, service, req_type='normal'):
"""Build the url to access the Boss' project service.
Args:
url_prefix (string): Do not end with a slash. Example of expected value: https://api.theboss.io
service (string): Name of service to access, such as collection, meta, coord, etc.
req_type (optional[string]): Valid values ['normal', 'list', 'cutout']. Defaults to 'normal'.
Returns:
(string): Full URL to access API.
Raises:
(RuntimeError): if invalid req_type given or url_prefix not given.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
if req_type == 'normal':
suffix = resource.get_route()
elif req_type == 'list':
# No suffix required for a list request.
suffix = resource.get_list_route()
elif req_type == 'cutout':
suffix = resource.get_cutout_route()
else:
raise RuntimeError('Invalid request type: {}'.format(req_type))
url = (url_prefix + '/' + self.version + '/' + service + '/' + suffix)
return url
def convert_int_list_range_to_str(
self, int_list)
Convert range in list of two ints to string representation.
Returned string can then be placed in URL as a query parameter.
Args:
int_list (list[int]): range such as [10, 20] which means x>=10 and x<20.
Returns
(string): [10, 20] => '10:20'
Raises
(RuntimeError): if given invalid range.
def convert_int_list_range_to_str(self, int_list):
"""Convert range in list of two ints to string representation.
Returned string can then be placed in URL as a query parameter.
Args:
int_list (list[int]): range such as [10, 20] which means x>=10 and x<20.
Returns:
(string): [10, 20] => '10:20'
Raises:
(RuntimeError): if given invalid range.
"""
if len(int_list) != 2:
raise RuntimeError('int_list must contain exactly two values.')
if int_list[0] > int_list[1]:
raise RuntimeError('Invalid range: int_list[0] > int_list[1].')
return '{}:{}'.format(int_list[0], int_list[1])
def convert_int_list_to_comma_sep_str(
self, int_list)
Convert list of ints to comma separated stringj.
Args:
int_list (list[int]): list of ints.
Returns
(string): Example: [1, 7, 9] => '1, 7, 9'
def convert_int_list_to_comma_sep_str(self, int_list):
"""Convert list of ints to comma separated stringj.
Args:
int_list (list[int]): list of ints.
Returns:
(string): Example: [1, 7, 9] => '1, 7, 9'
"""
if len(int_list) == 1:
return str(int_list[0])
str_list = []
for n in int_list:
str_list.append(str(n))
return (',').join(str_list)
def create(
self, resource, url_prefix, auth, session, send_opts)
Create the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises
requests.HTTPError on failure.
def create(self, resource, url_prefix, auth, session, send_opts):
"""Create the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises:
requests.HTTPError on failure.
"""
json = self._get_resource_params(resource)
req = self.get_request(resource, 'POST', 'application/json', url_prefix, auth, json=json)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return self._create_resource_from_dict(resource, resp.json())
err = ('Create failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def create_group(
self, name, url_prefix, auth, session, send_opts)
Create a new group.
Args:
name (string): Name of the group to create.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def create_group(self, name, url_prefix, auth, session, send_opts):
"""Create a new group.
Args:
name (string): Name of the group to create.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'POST', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 201:
return
msg = ('Create failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete(
self, resource, url_prefix, auth, session, send_opts)
Deletes the entity described by the given resource.
Args:
resource (intern.resource.boss.BossResource)
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def delete(self, resource, url_prefix, auth, session, send_opts):
"""Deletes the entity described by the given resource.
Args:
resource (intern.resource.boss.BossResource)
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'DELETE', 'application/json', url_prefix, auth)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
err = ('Delete failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def delete_group(
self, name, url_prefix, auth, session, send_opts)
Delete given group.
Args:
name (string): Name of group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def delete_group(self, name, url_prefix, auth, session, send_opts):
"""Delete given group.
Args:
name (string): Name of group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'DELETE', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Delete failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group_maintainer(
self, grp_name, user, url_prefix, auth, session, send_opts)
Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def delete_group_maintainer(
self, grp_name, user, url_prefix, auth, session, send_opts):
"""Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'DELETE', 'application/json', url_prefix, auth,
grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting maintainer {} from group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_group_member(
self, grp_name, user, url_prefix, auth, session, send_opts)
Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def delete_group_member(
self, grp_name, user, url_prefix, auth, session, send_opts):
"""Delete the given user from the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'DELETE', 'application/json', url_prefix, auth,
grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting maintainer {} from group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_permissions(
self, grp_name, resource, url_prefix, auth, session, send_opts)
Args:
grp_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def delete_permissions(self, grp_name, resource, url_prefix, auth, session, send_opts):
"""
Args:
grp_name (string): Name of group.
resource (intern.resource.boss.BossResource): Identifies which data model object to operate on.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
filter_params = {"group": grp_name}
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('DELETE', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = ('Failed deleting permissions to group {}, got HTTP response: ({}) - {}'.format(
grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request=req, response=resp)
def delete_user(
self, user, url_prefix, auth, session, send_opts)
Delete the given user.
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def delete_user(self, user, url_prefix, auth, session, send_opts):
"""Delete the given user.
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'DELETE', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = (
'Failed deleting user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def delete_user_role(
self, user, role, url_prefix, auth, session, send_opts)
Remove role from given user.
Args:
user (string): User name.
role (string): Role to remove.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises
requests.HTTPError on failure.
def delete_user_role(self, user, role, url_prefix, auth, session, send_opts):
"""Remove role from given user.
Args:
user (string): User name.
role (string): Role to remove.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_role_request(
'DELETE', 'application/json', url_prefix, auth,
user, role)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 204:
return
msg = (
'Failed deleting role: {} from user: {}, got HTTP response: ({}) - {}'
.format(role, user, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get(
self, resource, url_prefix, auth, session, send_opts)
Get attributes of the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises
requests.HTTPError on failure.
def get(self, resource, url_prefix, auth, session, send_opts):
"""Get attributes of the given resource.
Args:
resource (intern.resource.boss.BossResource): Create a data model object with attributes matching those of the resource.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(intern.resource.boss.BossResource): Returns resource of type requested on success.
Raises:
requests.HTTPError on failure.
"""
req = self.get_request(
resource, 'GET', 'application/json', url_prefix, auth)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return self._create_resource_from_dict(resource, resp.json())
err = ('Get failed on {}, got HTTP response: ({}) - {}'.format(
resource.name, resp.status_code, resp.text))
raise HTTPError(err, request = req, response = resp)
def get_bounding_box_request(
self, resource, method, content, url_prefix, token, resolution, id, bb_type)
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
resolution (int): 0 = default resolution.
id (int): Annotation object id.
bb_type (string): 'loose' | 'tight'.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_bounding_box_request(
self, resource, method, content, url_prefix, token, resolution,
id, bb_type):
"""
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
resolution (int): 0 = default resolution.
id (int): Annotation object id.
bb_type (string): 'loose' | 'tight'.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
url = (url_prefix + '/' + self.version + '/boundingbox/' +
resource.get_cutout_route() + '/{}/{}/?type={}'.format(
resolution, id, bb_type))
headers = self.get_headers(content, token)
return Request(method, url, headers=headers)
def get_cutout_request(
self, resource, method, content, url_prefix, token, resolution, x_range, y_range, z_range, time_range, numpyVolume=None, id_list=[])
Create a request for working with cutouts (part of the Boss' volume service).
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (list[int]): time range such as [30, 40] which means t>=30 and t<40.
numpyVolume (optional numpy array): The data volume encoded in a numpy array.
id_list (optional [list[int]]): list of object ids to filter the cutout by.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_cutout_request(
self, resource, method, content, url_prefix, token,
resolution, x_range, y_range, z_range, time_range, numpyVolume=None, id_list=[]):
"""Create a request for working with cutouts (part of the Boss' volume service).
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (list[int]): time range such as [30, 40] which means t>=30 and t<40.
numpyVolume (optional numpy array): The data volume encoded in a numpy array.
id_list (optional [list[int]]): list of object ids to filter the cutout by.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
url = self.build_cutout_url(
resource, url_prefix, resolution, x_range, y_range, z_range, time_range, id_list)
headers = self.get_headers(content, token)
return Request(method, url, headers = headers, data = numpyVolume)
def get_group(
self, name, user_name, url_prefix, auth, session, send_opts)
Get owner of group and the resources it's attached to.
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns
(dict): Keys include 'owner', 'name', 'resources'.
Raises
requests.HTTPError on failure.
def get_group(self, name, user_name, url_prefix, auth, session, send_opts):
"""Get owner of group and the resources it's attached to.
Args:
name (string): Name of group to query.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(dict): Keys include 'owner', 'name', 'resources'.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_request(
'GET', 'application/json', url_prefix, auth, name)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = ('Get failed for group {}, got HTTP response: ({}) - {}'.format(
name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_group_maintainers_request(
self, method, content, url_prefix, token, group_name, user_name=None)
Get a request object for working with group maintainers.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
group_name (string): Name of group.
user_name (optional[string]): Provide a user name if not doing a list operation. Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_group_maintainers_request(
self, method, content, url_prefix, token, group_name, user_name=None):
"""Get a request object for working with group maintainers.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
group_name (string): Name of group.
user_name (optional[string]): Provide a user name if not doing a list operation. Defaults to None.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
url = url_prefix + '/' + self.version + '/groups/' + group_name + '/maintainers/'
if user_name is not None:
url += user_name
headers = self.get_headers(content, token)
return Request(method, url, headers = headers)
def get_group_members_request(
self, method, content, url_prefix, token, group_name, user_name=None)
Get a request object for working with group membership.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
group_name (string): Name of group.
user_name (optional[string]): Provide a user name if not doing a list operation. Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_group_members_request(
self, method, content, url_prefix, token, group_name, user_name=None):
"""Get a request object for working with group membership.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
group_name (string): Name of group.
user_name (optional[string]): Provide a user name if not doing a list operation. Defaults to None.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
url = url_prefix + '/' + self.version + '/groups/' + group_name + '/members/'
if user_name is not None:
url += user_name
headers = self.get_headers(content, token)
return Request(method, url, headers = headers)
def get_group_request(
self, method, content, url_prefix, token, name=None)
Get a request for getting group information.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
name (optional[string|None]): Name of group. Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_group_request(self, method, content, url_prefix, token, name=None):
"""Get a request for getting group information.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
name (optional[string|None]): Name of group. Defaults to None.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
url = url_prefix + '/' + self.version + '/groups/'
if name is not None:
url += name + '/'
headers = self.get_headers(content, token)
return Request(method, url, headers = headers)
def get_headers(
self, content_type, token)
Get headers to place in Request object.
The auth token is added to the header for authenthication and
authorization.
Args:
content_type (string): HTTP content type of request.
token (string): Django Rest Framework token.
Returns
(dictionary): Use as headers argument to the Request object.
def get_headers(self, content_type, token):
"""Get headers to place in Request object.
The auth token is added to the header for authenthication and
authorization.
Args:
content_type (string): HTTP content type of request.
token (string): Django Rest Framework token.
Returns:
(dictionary): Use as headers argument to the Request object.
"""
return {
'Authorization': 'Token ' + token,
'Content-Type': content_type
}
def get_ids_request(
self, resource, method, content, url_prefix, token, resolution, x_range, y_range, z_range, time_range)
Create a request for getting ids in a region (part of the Boss' volume service).
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (list[int]): time range such as [30, 40] which means t>=30 and t<40.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_ids_request(
self, resource, method, content, url_prefix, token,
resolution, x_range, y_range, z_range, time_range):
"""Create a request for getting ids in a region (part of the Boss' volume service).
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
resolution (int): 0 indicates native resolution.
x_range (list[int]): x range such as [10, 20] which means x>=10 and x<20.
y_range (list[int]): y range such as [10, 20] which means y>=10 and y<20.
z_range (list[int]): z range such as [10, 20] which means z>=10 and z<20.
time_range (list[int]): time range such as [30, 40] which means t>=30 and t<40.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
url = self.build_ids_url(
resource, url_prefix, resolution, x_range, y_range, z_range, time_range)
headers = self.get_headers(content, token)
return Request(method, url, headers=headers)
def get_is_group_maintainer(
self, grp_name, user, url_prefix, auth, session, send_opts)
Check if the given user is a maintainer of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns
(bool): False if user not a maintainer.
Raises
requests.HTTPError on failure.
def get_is_group_maintainer(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Check if the given user is a maintainer of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(bool): False if user not a maintainer.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_maintainers_request(
'GET', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['result']
msg = ('Failed determining if user {} is maintainer of group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_is_group_member(
self, grp_name, user, url_prefix, auth, session, send_opts)
Check if the given user is a member of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns
(bool): False if user not a member.
Raises
requests.HTTPError on failure.
def get_is_group_member(self, grp_name, user, url_prefix, auth, session, send_opts):
"""Check if the given user is a member of the named group.
Args:
name (string): Name of group.
user (string): User to add to group.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(bool): False if user not a member.
Raises:
requests.HTTPError on failure.
"""
req = self.get_group_members_request(
'GET', 'application/json', url_prefix, auth, grp_name, user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
resp_json = resp.json()
return resp_json['result']
msg = ('Failed determining if user {} is member of group {}, got HTTP response: ({}) - {}'.format(
user, grp_name, resp.status_code, resp.text))
raise HTTPError(msg, request = req, response = resp)
def get_metadata_request(
self, resource, method, content, url_prefix, token, key=None, value=None)
Create a request for accessing the Boss' metadata service.
Do not use this method for list operations. Instead, use the
get_request() method.
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
key (string): Name of key to operate on. Defaults to None.
value (string): Value to assign to key Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_metadata_request(
self, resource, method, content, url_prefix, token,
key=None, value=None):
"""Create a request for accessing the Boss' metadata service.
Do not use this method for list operations. Instead, use the
get_request() method.
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
key (string): Name of key to operate on. Defaults to None.
value (string): Value to assign to key Defaults to None.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
url = self.build_metadata_url(resource, url_prefix, key, value)
headers = self.get_headers(content, token)
return Request(method, url, headers = headers)
def get_permission_request(
self, method, content, url_prefix, token, query_params=None, post_data=None)
Generate a request for manipulating permissions of a data model object.
Manipulate what members of the named group can do with the given data
model object.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
query_params (dict): Query params for GET requests. Defaults to None.
post_data (dict): POST body data. Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_permission_request(self, method, content, url_prefix, token, query_params=None, post_data=None):
"""Generate a request for manipulating permissions of a data model object.
Manipulate what members of the named group can do with the given data
model object.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
query_params (dict): Query params for GET requests. Defaults to None.
post_data (dict): POST body data. Defaults to None.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
url = (url_prefix + '/' + self.version + '/permissions/')
headers = self.get_headers(content, token)
if method == "GET" or method == "DELETE":
return Request(method, url, headers=headers, params=query_params)
else:
# Assuming POST or PATCH
return Request(method, url, headers=headers, json=post_data)
def get_permissions(
self, group_name, resource, url_prefix, auth, session, send_opts)
Get the permission set for a specific group/resource combination
Args:
group_name (string): Name of group to filter on
resource (intern.resource.boss.BossResource): Identifies which data model object to filter on
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns
(list[str]): List of permissions
def get_permissions(self, group_name, resource,url_prefix, auth, session, send_opts):
"""Get the permission set for a specific group/resource combination
Args:
group_name (string): Name of group to filter on
resource (intern.resource.boss.BossResource): Identifies which data model object to filter on
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(list[str]): List of permissions
"""
filter_params = {"group": group_name}
filter_params.update(resource.get_dict_route())
req = self.get_permission_request('GET', 'application/json',
url_prefix, auth, query_params=filter_params)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code != 200:
msg = "Failed to get permission set for Group: {} Resource: {}".format(group_name, resource.name)
msg = '{}, got HTTP response: ({}) - {}'.format(msg, resp.status_code, resp.text)
raise HTTPError(msg, request=req, response=resp)
else:
if resp.json()["permission-sets"]:
return resp.json()["permission-sets"][0]['permissions']
else:
return []
def get_request(
self, resource, method, content, url_prefix, token, proj_list_req=False, json=None, data=None)
Create a request for accessing the Boss' data model services.
Use for the project service or listing keys via the metadata service.
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
proj_list_req (bool): Set to True if performing a list operation on the project service. Defaults to False.
json (dict): POST body data. Defaults to None.
data (): Raw data. Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_request(self, resource, method, content, url_prefix, token, proj_list_req=False, json=None, data=None):
"""Create a request for accessing the Boss' data model services.
Use for the project service or listing keys via the metadata service.
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
proj_list_req (bool): Set to True if performing a list operation on the project service. Defaults to False.
json (dict): POST body data. Defaults to None.
data (): Raw data. Defaults to None.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if isinstance(resource, CoordinateFrameResource):
service = 'coord'
else:
service = 'collection'
if proj_list_req:
req_type = 'list'
else:
req_type = 'normal'
url = self.build_url(resource, url_prefix, service, req_type)
headers = self.get_headers(content, token)
req = Request(method, url, headers = headers, json = json, data = data)
return req
def get_reserve_request(
self, resource, method, content, url_prefix, token, num_ids)
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
num_ids (int): Number of ids to reserve.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_reserve_request(
self, resource, method, content, url_prefix, token, num_ids):
"""
Args:
resource (intern.resource.boss.BossResource): Resource to perform operation on.
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
num_ids (int): Number of ids to reserve.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
url = (url_prefix + '/' + self.version + '/reserve/' +
resource.get_reserve_route() + '/{}'.format(num_ids))
headers = self.get_headers(content, token)
return Request(method, url, headers=headers)
def get_user(
self, user, url_prefix, auth, session, send_opts)
Get user's data (first and last name, email, etc).
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns
(dictionary): User's data encoded in a dictionary.
Raises
requests.HTTPError on failure.
def get_user(self, user, url_prefix, auth, session, send_opts):
"""Get user's data (first and last name, email, etc).
Args:
user (string): User name.
url_prefix (string): Protocol + host such as https://api.theboss.io
auth (string): Token to send in the request header.
session (requests.Session): HTTP session to use for request.
send_opts (dictionary): Additional arguments to pass to session.send().
Returns:
(dictionary): User's data encoded in a dictionary.
Raises:
requests.HTTPError on failure.
"""
req = self.get_user_request(
'GET', 'application/json', url_prefix, auth,
user)
prep = session.prepare_request(req)
resp = session.send(prep, **send_opts)
if resp.status_code == 200:
return resp.json()
msg = (
'Failed getting user: {}, got HTTP response: ({}) - {}'
.format(user, resp.status_code, resp.text))
raise HTTPError(msg, request=req, response=resp)
def get_user_request(
self, method, content, url_prefix, token, user, first_name=None, last_name=None, email=None, password=None)
Generate a request for working with the /sso endpoint.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
user (string): Name of user.
first_name (optional[string]): User's first name. Defaults to None.
last_name (optional[string]): User's last name. Defaults to None.
email: (optional[string]): User's email address. Defaults to None.
password: (optional[string]): User's password. Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_user_request(
self, method, content, url_prefix, token, user, first_name=None,
last_name=None, email=None, password=None):
"""Generate a request for working with the /sso endpoint.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
user (string): Name of user.
first_name (optional[string]): User's first name. Defaults to None.
last_name (optional[string]): User's last name. Defaults to None.
email: (optional[string]): User's email address. Defaults to None.
password: (optional[string]): User's password. Defaults to None.
Returns:
(requests.Request): A newly constructed Request object.
Raises:
RuntimeError if url_prefix is None or an empty string.
"""
if url_prefix is None or url_prefix == '':
raise RuntimeError('url_prefix required.')
data = {}
url = url_prefix + '/' + self.version + '/sso/user/' + user
if first_name is not None:
data['first_name'] = first_name
if last_name is not None:
data['last_name'] = last_name
if email is not None:
data['email'] = email
if password is not None:
data['password'] = password
headers = self.get_headers(content, token)
return Request(method, url, headers=headers, json=data)
def get_user_role_request(
self, method, content, url_prefix, token, user, role=None)
Generate a request for manipulating roles for a user.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
user (string): Name of user.
role (optional[string]): Name of role. Defaults to None.
Returns
(requests.Request): A newly constructed Request object.
Raises
RuntimeError if url_prefix is None or an empty string.
def get_user_role_request(
self, method, content, url_prefix, token, user, role=None):
"""Generate a request for manipulating roles for a user.
Args:
method (string): HTTP verb such as 'GET'.
content (string): HTTP Content-Type such as 'application/json'.
url_prefix (string): protocol + initial portion of URL such as https://api.theboss.io Do not end with a forward slash.
token (string): Django Rest Framework token for auth.
user (string): Name of user.