ingest-accessioner/ingestapi.py at master · HumanCellAtlas/ingest-accessioner · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import json
import logging
import requests
import time
import urllib
import urlparse

ENTITY_TYPE_LINKS = {
    "sample": "samples",
    "assay": "assays",
    "analysis": "analyses",
    "file": "files",
    "project": "projects",
    "protocol": "protocols"
}

SEARCH_UUID_PATH = '/search/findByUuid?uuid='
MAX_PATCH_RETRIES = 3

LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
              '-35s %(lineno) -5d: %(message)s')
logging.basicConfig(level=logging.INFO)


class IngestApi:
    def __init__(self, ingest_url=None):
        reply = urllib.urlopen(ingest_url)
        self.links = json.load(reply)['_links']
        self.ingest_url = ingest_url

        self.logger = logging.getLogger(__name__)
        self.headers = {'Content-type': 'application/json', 'Accept': 'application/json'}

    def update_entity_by_uuid(self, entity_type, uuid, json_str):
        entity_url = self.get_entity_url_by_uuid(entity_type, uuid)
        r = requests.patch(entity_url, data=json_str, headers=self.headers)

        if r.status_code != requests.codes.ok:
            self.logger.error(str(r))

    def get_entity_url_by_uuid(self, entity_type, uuid):
        entity_index_url = self.get_entity_index(entity_type)
        entity_find_by_uuid_url = entity_index_url + SEARCH_UUID_PATH + uuid
        entity_response = urllib.urlopen(entity_find_by_uuid_url)
        entity_url = json.load(entity_response)['_links']['self']['href']

        return entity_url

    def get_entity_index_url(self, entity_type):
        metadata_type = ENTITY_TYPE_LINKS[entity_type.lower()]
        entity_index_url = self.links[metadata_type]['href'].rsplit('{')[0]
        return entity_index_url

    def update_entity(self, entity_type, entity_id, json_str):
        entity_url = self.get_entity_index_url(entity_type) + '/' + entity_id
        r = requests.patch(entity_url, data=json_str, headers=self.headers)

        if r.status_code != requests.codes.ok:
            self.logger.error(str(r))
        else:
            self.logger.info(str(r))

    def update_entity_if_match(self, entity_path, json_str):
        updated = False
        retries = 0

        while retries < MAX_PATCH_RETRIES:
            entity_url = urlparse.urljoin(self.ingest_url, entity_path)
            entity_response = requests.get(entity_url)
            etag = entity_response.headers['ETag']

            # recheck the response doesn't have a uuid
            metadata_uuid = json.loads(entity_response.text)['uuid']
            if not metadata_uuid:
                if etag:
                    self.headers['If-Match'] = etag

                self.logger.debug(self.headers)

                time.sleep(1)
                entity_update_response = requests.patch(entity_url, data=json_str, headers=self.headers)

                if entity_update_response.status_code != requests.codes.ok:
                    self.logger.error(str(entity_update_response))
                    retries += 1
                    self.logger.info('retries: ' + str(retries))
                else:
                    updated = True
                    break
            else:
                self.logger.info('Target document ' + str(entity_path) + ' already has UUID, ignoring')
                updated = False
                break
        return updated