Source code for gavo.formats.jsontable

"""
Writing tables in JSON.

We use python's built-in json engine to write an -- as yet -- ad-hoc json
format that essentially looks like this::

	{
		"contains": "table",
		"columns": { (column metadata more or less as in VOTable) }
		"data": { (rows as tuples) }
		("warnings": [...])
	}

No streaming at all is forseen for this format at this point.

Note that we don't do any custom type serialisers here; all the
stringification (or whatever) necessary happens within our own
mappers (utils.serializers or wherever something is added to the
defaultMFRegistry).
"""

#c Copyright 2008-2023, the GAVO project <gavo@ari.uni-heidelberg.de>
#c
#c This program is free software, covered by the GNU GPL.  See the
#c COPYING file in the source distribution.


import codecs
import json

from gavo import base
from gavo import rsc
from gavo.formats import common


[docs]class JSONMetaBuilder(base.MetaBuilder): """A MetaBuilder for mapping table meta information into our standard JSON structure. """ def __init__(self, jsonStructure): self.result = jsonStructure base.MetaBuilder.__init__(self)
[docs] def getResult(self): return self.result
[docs] def enterValue(self, value): if self.curAtoms[-1]=="_warning": self.result.setdefault("warnings", []).append(str(value)) elif self.curAtoms[-1]=="_queryStatus": self.result["queryStatus"] = str(value)
def _getJSONColumns(serManager): """returns a sequence of VOTable-like column description dictionaries. """ res = [] for annCol in serManager: res.append(annCol.annotations.copy()) res[-1].pop("displayHint", None) res[-1].pop("winningFactory", None) res[-1].pop("nullvalue", None) res[-1].pop("note", None) return res def _getJSONParams(serManager): """returns a dict of param dicts. """ result = { 'contains': "params", } for param in serManager.table.iterParams(): if param.value is not None: result[param.name] = { 'value': param.value, 'unit': param.unit, 'ucd': param.ucd, 'description': param.description,} return result def _getJSONStructure(table, acquireSamples=False): """returns a dictionary representing table for JSON serialisation. """ if isinstance(table, rsc.Data): table = table.getPrimaryTable() sm = base.SerManager(table, acquireSamples=acquireSamples) result = { 'contains': "table", 'params': _getJSONParams(sm), 'columns': _getJSONColumns(sm), 'data': list(sm.getMappedTuples()), } table.traverse(JSONMetaBuilder(result)) return result
[docs]def writeTableAsJSON(table, target, acquireSamples=False): """writes table to the target in ad-hoc JSON. """ jsonPayload = _getJSONStructure(table, acquireSamples) # our targets are always binary; wrap them for json.dump target = codecs.getwriter("utf-8")(target) return json.dump(jsonPayload, target)
# NOTE: while json could easily serialize full data elements, # right now we're only writing single tables. common.registerDataWriter("json", writeTableAsJSON, "application/json", "JSON", ".json")