from __future__ import print_function import urllib.request import argparse import json import datetime # This program shows examples of using the SRL API for modifying Resource and Collection data. # The API is meant to be discoverable, both by programs and people. # READ ALL NOTES BELOW BEFORE USING THE API # Please contact srl@imsweb.com if you require any assistance. # # SPECIFIC NOTES: # > Authentication # Authentication is token-based. First obtain a token by providing a username and password. # Subsequent requests provide the token as a HTTP request header. # # > Preventing Data Loss # When updating a Resource or Collection using the API, the Resource or Collection will be entirely # replaced by the update data. No partial updating of either a Resource or Collection is possible. # For example... to update the "organ_site" field for a Collection, all Collection data must # be provided, or the excluded field values will be empty after the update occurs. # To modify a single field, it is recommended to first retrieve the Resource or Collection # using the API, and then modify the retrieved data. # Resources MUST be updated INCLUDING ALL MEMBER COLLECTIONS or the collections will be DELETED. # # > Discovering the API # Navigate to https://specimens.cancer.gov/api to view a browsable version of the API. # Related items are linked to each other. # Click the 'OPTIONS' button to view the fields and HTTP request types that are supported # for each URL in the API. # # > "Staging" status # Brand new Resources and Collections are "staged" and must be approved by a site administrator # before they will appear on the site. Previously added Resources and Collections may be modified # using the API and their updates will appear live immediately. # SERVER_URL = 'https://specimens.cancer.gov' # This is the LIVE PRODUCTION site. SERVER_URL = 'https://www510.imsweb.com' # This is the BETA site TOKEN_GENERATING_URL = SERVER_URL + '/api/token-auth/' RESOURCE_LIST_CREATE_URL = SERVER_URL + '/api/resource/' COLLECTION_LIST_CREATE_URL = SERVER_URL + '/api/collection/' RESPONSE_OK = 200 RESPONSE_CREATED = 201 RESPONSE_DELETED = 204 def get_test_collection_data(): return { "name": "Test Collection", "organ_site": [ # valid values are here: https://specimens.cancer.gov/backend/codelists/#organ-site "Adrenal Glands" ], "tumor_type": [ # valid values are here: https://specimens.cancer.gov/backend/codelists/#histology-tumor-type "Benign Tumor", "Malignant Tumor", "Other (Benign Tumor)" ], "specimen_type": [ # valid values are here: https://specimens.cancer.gov/backend/codelists/#specimen-type "Surgical Resection" ], "secondary_specimen_types": [], # valid values are here: https://specimens.cancer.gov/backend/codelists/#specimen-type "preservation_type": [ # valid values are here: https://specimens.cancer.gov/backend/codelists/#preservation-type "Formalin-Fixed Paraffin-Embedded", "Fresh/Frozen" ], "available_data": [ # valid values are here: https://specimens.cancer.gov/backend/codelists/#available-data "Treatment Information", "Outcome Information", "Demographic Information", "Pathology Report" ], "number_of_specimens": 3, "collection_types": [ # valid values are here: https://specimens.cancer.gov/backend/codelists/#collection-type "Cancer Center", "Collaboration", "Institutional", "Pathology Department" ], "eligibility": [ # valid values are here: https://specimens.cancer.gov/backend/codelists/#who-is-eligible-to-apply "Non-U.S. Organization", "Non-Profit", "Academia", "Commercial", "Government" ], "is_collaboration_required": False, "clinical_trial_url": "" } def get_test_resource_data(): # Resources and Collections must have unique names. collection_x = get_test_collection_data() collection_x['name'] = collection_x['name'] + " X" collection_y = get_test_collection_data() collection_y['name'] = collection_y['name'] + " Y" return { "name": "Test Resource " + str(datetime.datetime.now()), "contact": "John Doe", "title": "Test coordinator", "phone": "999-999-9999", "email": "jdoe@test.com", "address": "Calverton, MD", "fax": "", "url": "www.imsweb.com", "service_shipping_fee": True, "publications": [], "collections": [collection_x, collection_y], # collections can be added at the same time as the resource. More than one collection may be added. "type": "Other" # valid values are here: https://specimens.cancer.gov/api/resource-type/ } def get_token(username, password): headers = {"Content-type": "application/json"} data = {'username': username, 'password': password} req = urllib.request.Request(TOKEN_GENERATING_URL, json.dumps(data), headers) response_code, response_dict = get_response(req) if response_code == RESPONSE_OK: print('The Token was created successfully.') else: print('The Token was not created successfully.') return response_dict.get('token', '') def get_resource_options(headers): request = urllib.request.Request(RESOURCE_LIST_CREATE_URL, json.dumps({}), headers=headers) request.get_method = lambda: 'OPTIONS' response_code, response_dict = get_response(request) if response_code == RESPONSE_OK: print('The Resource options were read successfully.') else: print('The Resource options were not read successfully.') return response_code, response_dict def get_collection_options(headers): request = urllib.request.Request(COLLECTION_LIST_CREATE_URL, json.dumps({}), headers=headers) request.get_method = lambda: 'OPTIONS' response_code, response_dict = get_response(request) if response_code == RESPONSE_OK: print('The Collection options were read successfully.') else: print('The Collection options were not read successfully.') return response_code, response_dict def create_new_resource(headers, data): request = urllib.request.Request(RESOURCE_LIST_CREATE_URL, json.dumps(data), headers=headers) response_code, response_dict = get_response(request) if response_code == RESPONSE_CREATED: resource_url = response_dict.get('id', '') print('The Resource was created successfully. It''s URL is %s.' % resource_url) else: print('The Resource was not created successfully.') return response_code, response_dict def update_resource(headers, resource_url, data): request = urllib.request.Request(resource_url, json.dumps(data), headers=headers) request.get_method = lambda: 'PUT' response_code, response_dict = get_response(request) if response_code == RESPONSE_OK: resource_url = response_dict.get('id', '') print('The Resource was updated successfully. It''s API URL is %s.' % resource_url) else: print('The Resource was not updated successfully.') return response_code, response_dict def update_collection(headers, collection_url, data): request = urllib.request.Request(collection_url, json.dumps(data), headers=headers) request.get_method = lambda: 'PUT' response_code, response_dict = get_response(request) if response_code == RESPONSE_OK: resource_url = response_dict.get('resource') collection_url = response_dict.get('id') print('The Collection was updated successfully. It''s API URL is %s. The resource to which the collection belongs has an API URL of %s.' % (collection_url, resource_url)) else: print('The Collection was not updated successfully.') return response_code, response_dict def get_resource(headers, resource_url): request = urllib.request.Request(resource_url, headers=headers) response_code, response_dict = get_response(request) if response_code == RESPONSE_OK: print('The Resource at %s was fetched successfully.' % resource_url) else: print('The Resource at %s was not fetched successfully.' % resource_url) return response_code, response_dict def get_collection(headers, collection_url): request = urllib.request.Request(collection_url, headers=headers) response_code, response_dict = get_response(request) if response_code == RESPONSE_OK: print('The Collection at %s was fetched successfully.' % collection_url) else: print('The Collection at %s was not fetched successfully.' % collection_url) return response_code, response_dict def delete_collection(headers, collection_url): request = urllib.request.Request(collection_url, json.dumps({}), headers=headers) request.get_method = lambda: 'DELETE' response_code, response_dict = get_response(request) if response_code == RESPONSE_DELETED: print('The Collection at %s was deleted successfully.' % collection_url) else: print('The Collection at %s was not deleted successfully.' % collection_url) return response_code, response_dict def delete_resource(headers, resource_url): request = urllib.request.Request(resource_url, json.dumps({}), headers=headers) request.get_method = lambda: 'DELETE' response_code, response_dict = get_response(request) if response_code == RESPONSE_DELETED: print('The Resource at %s was deleted successfully.' % resource_url) else: print('The Resource at %s was not deleted successfully.' % resource_url) return response_code, response_dict def get_response(request): response_code = None try: response = urllib.request.urlopen(request) response_data = response.read() response_code = response.code except urllib.request.HTTPError as error: response_data = error.read() response_code = error.code response_dict = json.loads(response_data) if response_data else {} return response_code, response_dict def print_resource_collections(resource_data): count = len(resource_data['collections']) names = ["'%s'" % collection['name'] for collection in resource_data['collections']] plural = 's' if count != 1 else '' name_str = ' and '.join(names) if count < 3 else '%s, and %s' % (','.join(names[0:-1]), names[-1]) print('The Resource currently has %d collection%s%s' % (count, plural, ": %s" % name_str if name_str else '')) def main(username, password): # All API methods require a token, so get this first. It can be re-used throughout this single session token = get_token(username, password) # Build the request headers. The "Authorization" header contains the token value, prepended with the string "JWT ". This exact header value formatting is required. headers = {"Content-type": "application/json", "Authorization": "JWT %s" % token } # Examine the field options for the JSON that the API expects response_code, resource_options = get_resource_options(headers) if response_code == RESPONSE_OK: print('Options for the "type" field of the Resource model, as reported by the OPTIONS request to %s:\n%s' % (RESOURCE_LIST_CREATE_URL, str(resource_options['actions']['POST']['type']))) response_code, collection_options = get_collection_options(headers) if response_code == RESPONSE_OK: print('Options for the "specimen_type" field of the Collection model, as reported by the OPTIONS request to %s:\n%s' % (COLLECTION_LIST_CREATE_URL, str(collection_options['actions']['POST']['specimen_type']))) # Create a new Resource resource_data = get_test_resource_data() response_code, new_resource_response_dict = create_new_resource(headers, resource_data) # Update the resource we just created by adding a Collection if response_code == RESPONSE_CREATED: resource_url = new_resource_response_dict['id'] # This is how to retrieve a fresh copy of the resource data using the API response_code, fresh_resource_data = get_resource(headers, resource_url) print_resource_collections(fresh_resource_data) # Using the retrieved data as a base, update Resource Name... fresh_resource_data['name'] = fresh_resource_data['name'] + ', with updates' # and then add a Collection to the Resource collection_a_data = get_test_collection_data() collection_a_data['name'] = collection_a_data['name'] + ' A' fresh_resource_data['collections'].append(collection_a_data) # Note that more than one Collection may be added, but collection name must be unique within the resource collection_b_data = get_test_collection_data() collection_b_data['name'] = collection_b_data['name'] + ' B' fresh_resource_data['collections'].append(collection_b_data) # The update operation also returns a copy of the Resource, after the updates are performed response_code, updated_resource_response_dict = update_resource(headers, resource_url, fresh_resource_data) if response_code == RESPONSE_OK: # If we wish to update the 1st collection, we can start by retrieving its data so that it can be used as a base. print_resource_collections(updated_resource_response_dict) first_collection_url = updated_resource_response_dict['collections'][0]['id'] if first_collection_url: response_code, fresh_collection_data = get_collection(headers, first_collection_url) if response_code == RESPONSE_OK: # make some changes to the collection data... fresh_collection_data['name'] = fresh_collection_data['name'] + ' updated name' fresh_collection_data['tumor_type'].pop() # ... and commit the changes. response_code, response_dict = update_collection(headers, first_collection_url, fresh_collection_data) if response_code == RESPONSE_OK: # Collections may be deleted. first_collection_url = response_dict['id'] response_code, response_dict = delete_collection(headers, first_collection_url) if response_code == RESPONSE_DELETED: # IMPORTANT: When updating a Resource or Collection, all fields must be provided at once. # If the collection element of the data POSTed for updating a Resource is empty, then all member collections will be deleted. updated_resource_response_dict['collections'] = [] response_code, updated_resource_without_any_collections = update_resource(headers, resource_url, updated_resource_response_dict) if response_code == RESPONSE_OK: # Since the Resource was updated with an empty 'collections' field, there are no member collections for the Resource anymore. print_resource_collections(updated_resource_without_any_collections) # Resources may also be deleted. response_code, response_dict = delete_resource(headers, resource_url) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument('username', type=str, help='User name') parser.add_argument('password', type=str, help='Password') args = parser.parse_args() main(**vars(args))