/
metadata.py
157 lines (133 loc) · 5.76 KB
/
metadata.py
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
import logging
from flask import request
from flask_restplus import Resource, marshal
from app.api.bigquery.serializers import table_response, column_response, substudy_response, study_response, study_response_test, table_response_short, unique_tissue_list, tissue_substudy
from app.api.bigquery.parsers import query_url_parser
from app.api.bigquery.business_metadata import *
from app.api.restplus import api
from app.database.helpers import populate_database
from app.database.models import *
from app import settings
from app.database import db
import requests
import json
log = logging.getLogger(__name__)
ns = api.namespace('metadata',
description="""Access the metadata for available datasets.
""")
@ns.route('/table')
class MetadataTableResources(Resource):
@ns.marshal_with(table_response_short)
def get(self):
"""Retrieve list of available tables"""
tables = get_table()
return tables, 200
@ns.response(404, "Table not found")
@ns.response(200, "OK")
@ns.doc(params={'table_name': 'the name of a biqquery table'})
@ns.route('/table/<string:table_name>')
class MetadataTableResource(Resource):
@ns.marshal_with(table_response)
def get(self, table_name):
"""Retrieve metadata about a table"""
try:
table = get_table(table_name)
return table, 200
except:
ns.abort(404, status='error', message="[%s] not a valid table" % (str(table_name), ))
@ns.response(404, "Column not found")
@ns.response(200, "OK")
@ns.doc(params={'table_name': 'the name of a biqquery table'})
@ns.doc(params={'column_name': 'the name of a column in this biqquery table'})
@ns.route('/table/<string:table_name>/column/<string:column_name>')
class MetadataColumnResource(Resource):
@ns.marshal_with(column_response)
def get(self, table_name, column_name):
"""Retrieve metadata about a tables column."""
try:
column = get_column(table_name, column_name)
return column, 200
except:
ns.abort(404, status='error', message="[%s.%s] not a valid column" % (str(table_name), str(column_name)))
@ns.route('/study')
class MetadataStudies(Resource):
@ns.marshal_with(study_response_test)
def get(self):
"""Return all available studies"""
return Study.query.all()
@ns.response(404, "Study not found")
@ns.response(200, "OK")
@ns.doc(params={'study_name': 'the name of a Big GIM study'})
@ns.route('/study/<string:study_name>')
class MetadataStudy(Resource):
@ns.marshal_with(study_response_test, code=200)
def get(self, study_name):
"""Return a single study and associated substudies"""
result =Study.query.filter_by(name=study_name).first()
if result:
return result, 200
else:
ns.abort(404, status='error', message="[%s] not a valid study" % (str(study_name),))
@ns.route('/tissue')
class Tissue(Resource):
@ns.marshal_with(unique_tissue_list)
def get(self):
"""Return a list of available tissues"""
tis = db.session.query(SubstudyTissue.tissue).distinct().all()
return {'tissues':sorted([t[0] for t in tis])}, 200
@ns.response(200, "OK")
@ns.response(404, "Tissue not found")
@ns.doc(params={'tissue_name': 'the name of tissue'})
@ns.route('/tissue/<string:tissue_name>')
class Tissue(Resource):
@ns.marshal_with(tissue_substudy)
def get(self, tissue_name):
"""Return a list of substudies and columns associated with a tissue"""
st = SubstudyTissue.query.filter_by(tissue=tissue_name).all()
if st:
substudies = []
for s in st:
substudies.append(Substudy.query.get(s.substudy_id))
return {'tissue':tissue_name, 'substudies':substudies}, 200
else:
ns.abort(404, status='error', message="[%s] not a valid tissue" % (str(tissue_name),))
@ns.route("/swagger")
class SwaggerSpec(Resource):
def get(self):
"""Return the swagger v2 spec for this API"""
return api.__schema__, 200
@ns.route("/openapiv3")
class OpenAPISpec(Resource):
def get(self):
"""Return the OpenAPI v3 spec for this API"""
import urllib
# use api coverter api to go from swagger -> openapi v3
swagger_path = "http://%s/api/metadata/%s" % (settings.FLASK_SERVER_NAME, 'swagger')
converter_path = "https://openapi-converter.herokuapp.com/api/v1/convert"
headers = {'accept':'application/json'}
full_path = "%s?%s" % (converter_path,
urllib.urlencode({'url':swagger_path}))
req = requests.get(full_path, headers=headers)
req.raise_for_status()
result = req.json()
# smartapi stuff
result['tags'] = [{'name':'translator'},{'name':'ISB'}, {'name':'gene'},
{'name':'similarity'},{'name':'tissue'}, {'name':'TCGA'},
{'name':'GIANT'}, {'name':'BioGrid'}, {'name':'GTEx'},
{'name':'BigQuery'}, {'name':'network'}, {'name': 'DeepTranslate'},
{'name':'query'} ]
result['info']['contact'] = {'name': 'John C. Earls', 'email':'john.c.earls@gmail.com'}
servers = [{'url':"http://%s/api" % (settings.FLASK_SERVER_NAME,)}]
result['servers'] = servers
return result, 200
@ns.route('/init_db/<string:password>', doc=False)
class Initdb(Resource):
def get(self, password):
"""This is a function that rebuilds the sqllite database"""
import hashlib
with open('/cred/database_reset.json') as pword_file:
hashed = json.load(pword_file)
if hashed['password'] == hashlib.sha224(password).hexdigest():
populate_database()
else:
ns.abort(401, status='error', message="Not a valid password")