In [12]:
def query_TAP(tap_endpoint, adql_query, table_to_upload=None):
"""
Query a TAP service (designated by its tap_endpoint)
with a given ADQL query
Query is performed synchronously
Return an AstroPy Table object
"""
import requests
from astropy.table import Table
from astropy.io.votable import parse_single_table
import os
import tempfile
import warnings
warnings.simplefilter("ignore")
r = requests.post(tap_endpoint + '/sync', data={'query': adql_query, 'request': 'doQuery', 'lang': 'adql', 'format': 'votable', 'phase': 'run'})
tmp_vot = tempfile.NamedTemporaryFile(delete = False)
with open(tmp_vot.name, 'w') as h:
for line in r.iter_lines():
if line:
h.write(line.decode(r.encoding)+'\n')
table = parse_single_table(tmp_vot.name).to_table()
# finally delete temp files
os.unlink(tmp_vot.name)
return table
In [18]:
source_id = 16527034310784
endpoint = 'http://tapvizier.u-strasbg.fr/TAPVizieR/tap'
adql = """SELECT ra, dec, pmra, pmdec, parallax, phot_g_mean_mag
FROM \"I/337/gaia\"
WHERE source_id = %d
""" % (source_id)
result = query_TAP(endpoint, adql)
print(result)