X
@@ -916,11 +857,7 @@ def danger_panel(var, warning): # to be used in create_var_vv
def info_panel(text, var='', id_var='', color_class='w3-sand'):
- # to print general info do not send var neither id_var
- # Newly created variant:
- c = 'c.'
- if re.search(r'N[MR]_', var):
- c = ''
+ c = '' if re.search(r'N[MR]_', var) else 'c.'
link = ''
if var != '':
link = """
@@ -964,30 +901,23 @@ def get_vv_api_url():
urls['variant_validator_api_hello_backup'],
urls['variant_validator_api_backup']
)
- if checked_url:
- return checked_url
- else:
+ if not checked_url:
checked_url = test_vv_api_url(
urls['variant_validator_api_hello'],
urls['variant_validator_api']
)
- return checked_url
+ return checked_url
else:
- # the other way around
- checked_url = test_vv_api_url(
- urls['variant_validator_api_hello'],
- urls['variant_validator_api']
- )
- if checked_url:
+ if checked_url := test_vv_api_url(
+ urls['variant_validator_api_hello'], urls['variant_validator_api']
+ ):
+ return checked_url
+ if checked_url := test_vv_api_url(
+ urls['variant_validator_api_hello_backup'],
+ urls['variant_validator_api_backup'],
+ ):
return checked_url
else:
- checked_url = test_vv_api_url(
- urls['variant_validator_api_hello_backup'],
- urls['variant_validator_api_backup']
- )
- if checked_url:
- return checked_url
- else:
send_error_email(
prepare_email_html(
'MobiDetails VariantValidator error',
diff --git a/MobiDetailsApp/upload.py b/MobiDetailsApp/upload.py
index 846151d5..e9246b06 100644
--- a/MobiDetailsApp/upload.py
+++ b/MobiDetailsApp/upload.py
@@ -41,149 +41,116 @@ def allowed_file(filename):
@bp.route('/file_upload', methods=['GET', 'POST'])
def file_upload():
- if request.method == 'POST':
- if request.files:
- uploaded_file = request.files['file']
- if uploaded_file.filename == "":
- flash('No filename.', 'w3-pale-red')
+ if request.method == 'POST' and request.files:
+ uploaded_file = request.files['file']
+ if uploaded_file.filename == "":
+ flash('No filename.', 'w3-pale-red')
+ return render_template('upload/upload_form.html')
+ # return redirect(request.url)
+ if allowed_file(uploaded_file.filename):
+ # filename = secure_filename(uploaded_file.filename)
+ lines = uploaded_file.read().decode().replace('\r\n', '\n').replace('\r', '\n').split('\n')
+ # print(lines)
+ # ok we've got the file
+ # get user API key or use MD
+ db = get_db()
+ curs = db.cursor(cursor_factory=psycopg2.extras.DictCursor)
+ http = urllib3.PoolManager(cert_reqs='CERT_REQUIRED', ca_certs=certifi.where())
+ # headers
+ header = md_utilities.api_agent
+ result = []
+ api_key = md_utilities.get_api_key(g, curs)
+ if api_key is None:
+ flash('There is an issue in obtaining an API key')
+ close_db()
return render_template('upload/upload_form.html')
# return redirect(request.url)
- if allowed_file(uploaded_file.filename):
- # filename = secure_filename(uploaded_file.filename)
- lines = uploaded_file.read().decode().replace('\r\n', '\n').replace('\r', '\n').split('\n')
- # print(lines)
- # ok we've got the file
- # get user API key or use MD
- db = get_db()
- curs = db.cursor(cursor_factory=psycopg2.extras.DictCursor)
- http = urllib3.PoolManager(cert_reqs='CERT_REQUIRED', ca_certs=certifi.where())
- # headers
- header = md_utilities.api_agent
- result = []
- api_key = md_utilities.get_api_key(g, curs)
- if api_key is None:
- flash('There is an issue in obtaining an API key')
- close_db()
- return render_template('upload/upload_form.html')
- # return redirect(request.url)
# we need to check the format and send a proper query to the API
- for line in lines:
- # print('-{}-'.format(line))
- # cDNA format
- md_response = []
- if re.search(r'^#', line) or \
- line == '':
- continue
- line = line.replace(' ', '')
- variant_regexp = md_utilities.regexp['variant']
- ncbi_transcript_regexp = md_utilities.regexp['ncbi_transcript']
- match_obj_c = re.search(rf'^({ncbi_transcript_regexp})\(*[\w-]*\)*:(c\.{variant_regexp})$', line)
- if match_obj_c:
+ for line in lines:
+ # print('-{}-'.format(line))
+ # cDNA format
+ md_response = []
+ if re.search(r'^#', line) or \
+ line == '':
+ continue
+ line = line.replace(' ', '')
+ variant_regexp = md_utilities.regexp['variant']
+ ncbi_transcript_regexp = md_utilities.regexp['ncbi_transcript']
+ if match_obj_c := re.search(
+ rf'^({ncbi_transcript_regexp})\(*[\w-]*\)*:(c\.{variant_regexp})$',
+ line,
+ ):
# check NM number and version
- curs.execute(
- """
+ curs.execute(
+ """
SELECT gene_symbol
FROM gene
WHERE refseq = %s
""",
- (match_obj_c.group(1),)
- )
- res_gene = curs.fetchone()
- if res_gene:
- # send var to api
- md_api_url = '{0}{1}'.format(request.host_url[:-1], url_for('api.api_variant_create'))
+ (match_obj_c[1],),
+ )
+ if res_gene := curs.fetchone():
+ # send var to api
+ md_api_url = '{0}{1}'.format(request.host_url[:-1], url_for('api.api_variant_create'))
# md_api_url = '{0}/api/variant/create'.format(md_api_base_url)
# print(md_api_url)
- data = {
- 'variant_chgvs': urllib.parse.quote(
- '{0}:{1}'.format(
- match_obj_c.group(1),
- match_obj_c.group(2)
- )
- ),
- 'caller': 'cli',
- 'api_key': api_key
- }
- try:
- md_response = json.loads(
- http.request(
- 'POST',
- md_api_url,
- headers=header,
- fields=data
- ).data.decode('utf-8')
- )
- result.append(
- {'variant': line,
- 'id': md_response['mobidetails_id'],
- 'url': md_response['url']}
+ data = {
+ 'variant_chgvs': urllib.parse.quote(
+ '{0}:{1}'.format(
+ match_obj_c[1], match_obj_c[2]
)
- except Exception:
- if 'mobidetails_error' in md_response:
- result.append({'variant': line, 'error': md_response['mobidetails_error']})
- else:
- result.append({'variant': line, 'error': 'MDAPI call failed'})
- else:
- result.append({'variant': line, 'error': 'Unknown NCBI NM accession number'})
- continue
- # genomic format
- ncbi_chrom_regexp = md_utilities.regexp['ncbi_chrom']
- match_obj_g = re.search(rf'^({ncbi_chrom_regexp}:g\.{variant_regexp});([\w-]+)$', line)
- if match_obj_g:
+ ),
+ 'caller': 'cli',
+ 'api_key': api_key,
+ }
+ try:
+ md_response = json.loads(
+ http.request(
+ 'POST',
+ md_api_url,
+ headers=header,
+ fields=data
+ ).data.decode('utf-8')
+ )
+ result.append(
+ {'variant': line,
+ 'id': md_response['mobidetails_id'],
+ 'url': md_response['url']}
+ )
+ except Exception:
+ if 'mobidetails_error' in md_response:
+ result.append({'variant': line, 'error': md_response['mobidetails_error']})
+ else:
+ result.append({'variant': line, 'error': 'MDAPI call failed'})
+ else:
+ result.append({'variant': line, 'error': 'Unknown NCBI NM accession number'})
+ continue
+ # genomic format
+ ncbi_chrom_regexp = md_utilities.regexp['ncbi_chrom']
+ if match_obj_g := re.search(
+ rf'^({ncbi_chrom_regexp}:g\.{variant_regexp});([\w-]+)$',
+ line,
+ ):
# check gene is available
- curs.execute(
- """
+ curs.execute(
+ """
SELECT gene_symbol
FROM gene
WHERE gene_symbol = %s
""",
- (match_obj_g.group(2),)
- )
- res_gene = curs.fetchone()
- if res_gene:
- # send var to api
- md_api_url = '{0}{1}'.format(request.host_url[:-1], url_for('api.api_variant_g_create'))
+ (match_obj_g[2],),
+ )
+ if res_gene := curs.fetchone():
+ # send var to api
+ md_api_url = '{0}{1}'.format(request.host_url[:-1], url_for('api.api_variant_g_create'))
# print(md_api_url)
- data = {
- 'variant_ghgvs': urllib.parse.quote(match_obj_g.group(1)),
- 'gene_hgnc': match_obj_g.group(2),
- 'caller': 'cli',
- 'api_key': api_key
- }
- try:
- md_response = json.loads(
- http.request(
- 'POST',
- md_api_url,
- headers=header,
- fields=data
- ).data.decode('utf-8')
- )
- # print(md_response)
- result.append({'variant': line, 'id': md_response['mobidetails_id']})
- except Exception:
- if 'variant_validator_output' in md_response and \
- 'validation_warning_1' in md_response['variant_validator_output'] and \
- 'validation_warnings' in md_response['variant_validator_output']['validation_warning_1']:
- result.append(
- {'variant': line,
- 'error': md_response['variant_validator_output']['validation_warning_1']['validation_warnings'][0]}
- )
- elif 'mobidetails_error' in md_response:
- result.append({'variant': line, 'error': md_response['mobidetails_error']})
- else:
- result.append({'variant': line, 'error': 'MDAPI call failed'})
- else:
- result.append({'variant': line, 'error': 'Unknown gene'})
- continue
- if re.search(r'^rs\d+$', line):
- md_api_url = '{0}{1}'.format(request.host_url[:-1], url_for('api.api_variant_create_rs'))
- # md_api_url = '{0}/api/variant/create'.format(md_api_base_url)
- # print(md_api_url)
data = {
- 'rs_id': line,
+ 'variant_ghgvs': urllib.parse.quote(
+ match_obj_g[1]
+ ),
+ 'gene_hgnc': match_obj_g[2],
'caller': 'cli',
- 'api_key': api_key
+ 'api_key': api_key,
}
try:
md_response = json.loads(
@@ -194,73 +161,108 @@ def file_upload():
fields=data
).data.decode('utf-8')
)
- for var in md_response:
- # print(md_response[var])
- if 'mobidetails_error' in md_response[var]:
- result.append(
- {'variant': '{0} - {1}'.format(line, var),
- 'error': md_response[var]['mobidetails_error']}
- )
- elif var == 'mobidetails_error':
- result.append({'variant': line, 'error': md_response[var]})
- else:
- result.append(
- {'variant': '{0} - {1}'.format(line, var),
- 'id': md_response[var]['mobidetails_id'],
- 'url': md_response[var]['url']}
- )
-
+ # print(md_response)
+ result.append({'variant': line, 'id': md_response['mobidetails_id']})
except Exception:
- result.append({'variant': line, 'error': 'MDAPI call failed'})
- continue
- result.append({'variant': line, 'error': 'Bad format'})
- close_db()
- flash('File correctly uploaded', 'w3-pale-green')
- if g.user:
- # send an email
- # print(g.user)
- result_list = ''
- for resul in result:
- result_list = '{0}
{1}'.format(result_list, resul['variant'])
- if 'error' in resul:
- result_list = '{0} - {1}'.format(result_list, resul['error'])
- else:
- result_list = '{0} -
success'.format(
- result_list,
- request.host_url[:-1],
- url_for(
- 'api.variant',
- variant_id=resul['id'],
- caller='browser'
+ if 'variant_validator_output' in md_response and \
+ 'validation_warning_1' in md_response['variant_validator_output'] and \
+ 'validation_warnings' in md_response['variant_validator_output']['validation_warning_1']:
+ result.append(
+ {'variant': line,
+ 'error': md_response['variant_validator_output']['validation_warning_1']['validation_warnings'][0]}
)
+ elif 'mobidetails_error' in md_response:
+ result.append({'variant': line, 'error': md_response['mobidetails_error']})
+ else:
+ result.append({'variant': line, 'error': 'MDAPI call failed'})
+ else:
+ result.append({'variant': line, 'error': 'Unknown gene'})
+ continue
+ if re.search(r'^rs\d+$', line):
+ md_api_url = '{0}{1}'.format(request.host_url[:-1], url_for('api.api_variant_create_rs'))
+ # md_api_url = '{0}/api/variant/create'.format(md_api_base_url)
+ # print(md_api_url)
+ data = {
+ 'rs_id': line,
+ 'caller': 'cli',
+ 'api_key': api_key
+ }
+ try:
+ md_response = json.loads(
+ http.request(
+ 'POST',
+ md_api_url,
+ headers=header,
+ fields=data
+ ).data.decode('utf-8')
+ )
+ for var in md_response:
+ # print(md_response[var])
+ if 'mobidetails_error' in md_response[var]:
+ result.append(
+ {'variant': '{0} - {1}'.format(line, var),
+ 'error': md_response[var]['mobidetails_error']}
+ )
+ elif var == 'mobidetails_error':
+ result.append({'variant': line, 'error': md_response[var]})
+ else:
+ result.append(
+ {'variant': '{0} - {1}'.format(line, var),
+ 'id': md_response[var]['mobidetails_id'],
+ 'url': md_response[var]['url']}
+ )
+
+ except Exception:
+ result.append({'variant': line, 'error': 'MDAPI call failed'})
+ continue
+ result.append({'variant': line, 'error': 'Bad format'})
+ close_db()
+ flash('File correctly uploaded', 'w3-pale-green')
+ if g.user:
+ # send an email
+ # print(g.user)
+ result_list = ''
+ for resul in result:
+ result_list = '{0}
{1}'.format(result_list, resul['variant'])
+ if 'error' in resul:
+ result_list = '{0} - {1}'.format(result_list, resul['error'])
+ else:
+ result_list = '{0} -
success'.format(
+ result_list,
+ request.host_url[:-1],
+ url_for(
+ 'api.variant',
+ variant_id=resul['id'],
+ caller='browser'
)
+ )
- message = """
+ message = """
Dear {0},
Your batch job returned the following results:
You can have a direct acces to the successfully annotated variants at your
profile page.
""".format(
- g.user['username'],
- result_list,
- request.host_url[:-1],
- url_for(
- 'auth.profile',
- mobiuser_id=0
- )
- )
- md_utilities.send_email(
- md_utilities.prepare_email_html(
- 'MobiDetails - Batch job',
- message,
- False
- ),
- '[MobiDetails - Batch job]',
- [g.user['email']]
+ g.user['username'],
+ result_list,
+ request.host_url[:-1],
+ url_for(
+ 'auth.profile',
+ mobiuser_id=0
)
- return render_template('md/variant_multiple.html', upload=result)
- else:
- flash('That file extension is not allowed', 'w3-pale-red')
- return render_template('upload/upload_form.html')
- # return redirect(request.url)
+ )
+ md_utilities.send_email(
+ md_utilities.prepare_email_html(
+ 'MobiDetails - Batch job',
+ message,
+ False
+ ),
+ '[MobiDetails - Batch job]',
+ [g.user['email']]
+ )
+ return render_template('md/variant_multiple.html', upload=result)
+ else:
+ flash('That file extension is not allowed', 'w3-pale-red')
+ return render_template('upload/upload_form.html')
+ # return redirect(request.url)
return render_template("upload/upload_form.html", run_mode=md_utilities.get_running_mode())
diff --git a/tests/test_ajax.py b/tests/test_ajax.py
index a40ad3cf..3c529f82 100644
--- a/tests/test_ajax.py
+++ b/tests/test_ajax.py
@@ -201,7 +201,13 @@ def test_lovd(client, app):
res = curs.fetchall()
db_pool.putconn(db)
for values in res:
- data_dict = dict(genome=values['genome_version'], chrom=values['chr'], g_name=values['g_name'], c_name='c.{}'.format(values['c_name']), gene=values['gene_symbol'])
+ data_dict = dict(
+ genome=values['genome_version'],
+ chrom=values['chr'],
+ g_name=values['g_name'],
+ c_name=f"c.{values['c_name']}",
+ gene=values['gene_symbol'],
+ )
print(data_dict)
assert client.post('/lovd', data=data_dict).status_code == 200
@@ -434,20 +440,23 @@ def test_delete_variant_list(client, app, auth, list_name, return_value):
assert client.get('/delete_variant_list').status_code == 404
with app.app_context():
response = client.get(
- '/delete_variant_list/{}'.format(list_name),
- follow_redirects=True
+ f'/delete_variant_list/{list_name}', follow_redirects=True
)
print(response.get_data())
assert b'check_login_form' in response.get_data() # means we are in the login page
auth.login(email, password)
- assert client.get(
- '/delete_variant_list/{}'.format(list_name),
- follow_redirects=True
- ).status_code == return_value
- assert b'check_login_form' in client.get(
- '/delete_variant_list/{}'.format(list_name),
- follow_redirects=True
- ).get_data()
+ assert (
+ client.get(
+ f'/delete_variant_list/{list_name}', follow_redirects=True
+ ).status_code
+ == return_value
+ )
+ assert (
+ b'check_login_form'
+ in client.get(
+ f'/delete_variant_list/{list_name}', follow_redirects=True
+ ).get_data()
+ )
# test autocomplete
diff --git a/tests/test_api.py b/tests/test_api.py
index f8ea9736..a1a73206 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -49,7 +49,9 @@ def test_check_vv_instance(client):
('C_000001.11:g.216422237G>A', 'mobidetails_error', 'Malformed query')
))
def test_api_variant_exists(client, variant, key, response):
- json_response = json.loads(client.get('/api/variant/exists/{}'.format(variant)).data.decode('utf8'))
+ json_response = json.loads(
+ client.get(f'/api/variant/exists/{variant}').data.decode('utf8')
+ )
assert response in str(json_response[key])
# test gene
@@ -62,7 +64,7 @@ def test_api_variant_exists(client, variant, key, response):
('--%USH2A', 'mobidetails_error', 'Invalid gene submitted')
))
def test_api_gene(client, gene, key, response):
- json_response = json.loads(client.get('/api/gene/{}'.format(gene)).data.decode('utf8'))
+ json_response = json.loads(client.get(f'/api/gene/{gene}').data.decode('utf8'))
print(json_response)
assert response in str(json_response[key])
@@ -93,7 +95,9 @@ def get_generic_api_key():
(8, 'hg19PseudoVCF', '1-216420460-C-A'),
))
def test_api_variant(client, variant_id, key, value):
- json_response = json.loads(client.get('/api/variant/{}/clispip/'.format(variant_id)).data.decode('utf8'))
+ json_response = json.loads(
+ client.get(f'/api/variant/{variant_id}/clispip/').data.decode('utf8')
+ )
assert json_response['nomenclatures'][key] == value
# 2nd test variant creation
@@ -116,7 +120,11 @@ def test_api_variant2(app, client):
db_pool.putconn(db)
for variant_id in res:
print(variant_id)
- json_response = json.loads(client.get('/api/variant/{}/cli/'.format(variant_id[0])).data.decode('utf8'))
+ json_response = json.loads(
+ client.get(f'/api/variant/{variant_id[0]}/cli/').data.decode(
+ 'utf8'
+ )
+ )
assert 'nomenclatures' in json_response
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 81db7051..6e1d05a7 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -96,9 +96,9 @@ def test_activate(client, mobiuser_id, api_key, message):
(10, b'Username'),
))
def test_profile(client, app, auth, mobiuser_id, message):
- assert client.get('/auth/profile/{}'.format(mobiuser_id)).status_code == 302
+ assert client.get(f'/auth/profile/{mobiuser_id}').status_code == 302
with app.app_context():
- response = client.get('/auth/profile/{}'.format(mobiuser_id), follow_redirects=True)
+ response = client.get(f'/auth/profile/{mobiuser_id}', follow_redirects=True)
assert b'check_login_form' in response.get_data() # means we are in the login page
# following does not work - as if login does not work properly?
# db_pool, db = get_db()
@@ -169,8 +169,4 @@ def test_reset_password(client, mobiuser_id, api_key, timestamp, message):
('test', 404),
))
def test_variant_list(client, list_name, http_code):
- assert client.get(
- '/auth/variant_list/{}'.format(
- list_name,
- )
- ).status_code == http_code
+ assert client.get(f'/auth/variant_list/{list_name}').status_code == http_code
diff --git a/tests/test_md.py b/tests/test_md.py
index c7499461..0384a144 100644
--- a/tests/test_md.py
+++ b/tests/test_md.py
@@ -25,7 +25,7 @@ def test_gene_page(client, app):
db_pool.putconn(db)
for name in res:
print(name[0])
- assert client.get('/gene/{}'.format(name[0])).status_code == 200
+ assert client.get(f'/gene/{name[0]}').status_code == 200
def test_genes_page(client):
@@ -97,8 +97,8 @@ def test_search_engine(client, app, t_search, url, status_code):
print(response.get_data())
assert url in response.get_data()
elif status_code == 302:
- print(response.headers['Location'] + 'http://localhost/{}'.format(url))
- assert 'http://localhost/{}'.format(url) == response.headers['Location']
+ print(response.headers['Location'] + f'http://localhost/{url}')
+ assert f'http://localhost/{url}' == response.headers['Location']
else:
assert status_code == response.status_code
@@ -123,4 +123,4 @@ def test_variant_page(client, app):
db_pool.putconn(db)
for variant_id in res:
print(variant_id[0])
- assert client.get('/api/variant/{}/browser/'.format(variant_id[0])).status_code == 200
+ assert client.get(f'/api/variant/{variant_id[0]}/browser/').status_code == 200
diff --git a/tests/test_md_utilities.py b/tests/test_md_utilities.py
index e52bf607..5f078cbb 100644
--- a/tests/test_md_utilities.py
+++ b/tests/test_md_utilities.py
@@ -960,7 +960,7 @@ def test_lovd_error_html():
))
def test_format_mirs(record, result):
totest = md_utilities.format_mirs(record)
- print('-{}'.format(totest))
+ print(f'-{totest}')
assert re.split(r'\s+', result) == re.split(r'\s+', totest)