"""
Support for IVOA DAL and registry protocols.
DALI-specified protocols are in dalirender, UWS/async is in asyncrender.
"""
#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 datetime
import os
from twisted.internet import defer
from twisted.web import server
from gavo import base
from gavo import formats
from gavo import registry
from gavo import rscdef
from gavo import rsc
from gavo import svcs
from gavo import utils
from gavo import votable
from gavo.protocols import dali
from gavo.protocols import soda
from gavo.protocols import ssap
from gavo.svcs import streaming
from gavo.votable import V
from gavo.web import grend
from gavo.web import weberrors
MS = base.makeStruct
[docs]class DALRenderer(grend.ServiceBasedPage):
"""A base class for renderers for the usual IVOA DAL protocols.
This is for simple, GET-based DAL renderers (where we allow POST as
well). They work using nevow forms, but with standard-compliant error
reporting (i.e., in VOTables).
Since DALRenderer mixes in FormMixin, it always has the form genFrom.
"""
resultType = base.votableType
urlUse = "base"
standardId = None
defaultOutputFormat = "votable"
def __init__(self, request, *args, **kwargs):
grend.ServiceBasedPage.__init__(self, request, *args, **kwargs)
self.defaultLimit = base.getConfig("ivoa", "dalDefaultLimit")
[docs] @classmethod
def makeAccessURL(cls, baseURL):
return "%s/%s?"%(baseURL, cls.name)
[docs] @classmethod
def isBrowseable(self, service):
return False
[docs] def render(self, request, sync=False):
"""runs the service and arranges for output to be produced.
This will in general result in a deferred, except if you pass sync=True,
in which case the thing tries to run everything in one go.
Don't do that outside of unit tests.
"""
if sync:
runner = self.runSync
else:
runner = self.runAsync
# overwrite previous query meta for new limit
self.queryMeta = svcs.QueryMeta.fromRequest(request,
defaultLimit=self.defaultLimit)
def execRunner():
if (self.queryMeta["dbLimit"]==0
or self.queryMeta.ctxArgs.get("FORMAT", "").lower()=="metadata"):
return self._produceMetadata(request)
else:
dali.mangleUploads(request)
return runner(request.strargs)
defer.maybeDeferred(execRunner
).addCallback(self._formatOutput, request, stream=not sync
).addErrback(self._handleInputErrors, request
).addErrback(self._handleRandomFailure, request
).addErrback(request.finishCallback)
return server.NOT_DONE_YET
[docs] def renderSync(self, request):
# This does essentially what render does, but synchronously.
# This *should* work for DAL protocols, but it's really only
# intended for unit testing. It's not a bug in a renderer or
# core if renderSync doesn't work for it.
# Also, this only has sense with a FakeRequest, as that's where
# we take our result data from.
res = self.render(request, sync=True)
if res==server.NOT_DONE_YET:
# well, we assume things really ran sync and let it go.
pass
elif res:
request.write(res)
return request.accumulator
def _getMetadataData(self):
"""returns a SIAP-style metadata data item.
"""
inputFields = []
for param in self.service.getInputKeysFor(self):
# Caution: UPLOAD mangling is a *renderer* thing -- the service
# doesn't know anything about it. Hence, parameter adaption
# is *not* done by adapting the real service metadata. Instead:
if param.type=="file":
param = dali.getUploadKeyFor(param)
inputFields.append(param.change(name="INPUT:"+param.name))
outputTD = self.service.core.outputTable.change(id="results")
for param in outputTD.params:
param.name = "OUTPUT:"+param.name
nullRowmaker = MS(rscdef.RowmakerDef)
dataDesc = MS(rscdef.DataDescriptor, makes=[
MS(rscdef.Make, table=outputTD, rowmaker=nullRowmaker)],
params=inputFields,
parent_=self.service.rd)
data = rsc.makeData(dataDesc)
data.contributingMetaCarriers.append(self.service)
data.tables["results"].votCasts = self._outputTableCasts
data.setMeta("_type", "meta")
data.addMeta("info",
base.getMetaText(self.service, "title") or "Unnamed",
infoName="serviceInfo",
infoValue=str(self.service.getURL(self.name)))
return data
def _produceMetadata(self, request):
metaData = self._getMetadataData()
metaData.addMeta("info", "OK",
infoName="QUERY_STATUS", infoValue="OK")
request.setHeader("content-type", "text/xml")
votLit = formats.getFormatted("votable", metaData)
# maybe provide a better way to attach stylesheet info?
splitPos = votLit.find(b"?>")+2
return base.votableType, (votLit[:splitPos]+(
b"<?xml-stylesheet href='/static"
b"/xsl/meta-votable-to-html.xsl' type='text/xsl'?>"
)+votLit[splitPos:])
def _writeErrorTable(self,
request,
errmsg,
httpStatus=200,
queryStatus="ERROR"):
# Unfortunately, most legacy DAL specs say the error messages must
# be delivered with a 200 response code. I hope this is going
# to change at some point, so I let renderers order sane response
# codes.
if not request.client:
# remote side has gone away -- avoid triggering ugly errors
return
return dali.serveDALIError(request, errmsg, httpStatus, queryStatus)
def _formatOutput(self, data, request, stream=True):
if isinstance(data, tuple):
# core returned a complete document (mime and string)
mime, payload = data
request.setHeader("content-type", mime)
if stream:
return streaming.streamOut(
lambda f: f.write(payload), request, self.queryMeta)
else:
request.write(payload)
request.finish()
return server.NOT_DONE_YET
data.contributingMetaCarriers.append(self.service)
data.addMeta("info", "", infoName="QUERY_STATUS",
infoValue=base.getMetaText(data.getPrimaryTable(),
"_queryStatus"))
data.addMeta("info", "", infoName="request",
infoValue=utils.debytify(request.uri))
if self.standardId:
data.addMeta("info",
"Written by DaCHS %s %s"%(base.getVersion(), self.__class__.__name__),
infoName="standardID",
infoValue=self.standardId)
destFormat = self.defaultOutputFormat
if "responseformat" in self.queryMeta.ctxArgs:
# This is our DALI RESPONSEFORMAT implementation; the corresponding
# parameter is declared automatically in Service.completeElement
# from pql#DALIPars.
destFormat = self.queryMeta.ctxArgs["responseformat"]
requestedType = formats.getMIMEFor(destFormat, destFormat)
else:
requestedType = self.resultType
request.setHeader("content-type", requestedType)
request.setHeader('content-disposition',
'attachment; filename="result.%s"'%
formats.getExtensionFor(requestedType))
formatterArgs = {}
if stream:
def writeStuff(outputFile):
formats.formatData(destFormat,
data, outputFile, acquireSamples=False, **formatterArgs)
return streaming.streamOut(writeStuff, request, self.queryMeta)
else:
return formats.formatData(destFormat,
data, request, acquireSamples=False, **formatterArgs)
def _handleRandomFailure(self, failure, request):
base.ui.notifyFailure(failure)
return self._writeErrorTable(request, failure.value, 500)
def _handleInputErrors(self, failure, request):
httpStatus, queryStatus = 200, "ERROR"
if base.DEBUG:
base.ui.notifyFailure(failure)
if isinstance(failure.value, base.EmptyData):
httpStatus, queryStatus = 400, "EMPTY"
return self._writeErrorTable(
request,
failure.value,
httpStatus=httpStatus,
queryStatus=queryStatus)
[docs]class SCSRenderer(DALRenderer):
"""
A renderer for the Simple Cone Search protocol.
These do their error signaling in the value attribute of an
INFO child of RESOURCE.
You must set the following metadata items on services using
this renderer if you want to register them:
* testQuery.ra, testQuery.dec -- A position for which an object is present
within 0.001 degrees.
"""
name = "scs.xml"
version = "1.0"
parameterStyle = "dali"
standardId = "ivo://ivoa.net/std/ConeSearch"
# move the ucdCasts from formatOutput here.
_outputTableCasts = {}
def __init__(self, request, *args, **kwargs):
DALRenderer.__init__(self, request, *args, **kwargs)
self.defaultLimit = base.getConfig("ivoa", "dalDefaultLimit")*10
def _writeErrorTable(self, request, msg, httpStatus=200, queryStatus="ERROR"):
if not request.client:
# remote side has gone away -- avoid triggering ugly errors
return
request.setHeader("content-type", base.votableType)
votable.write(V.VOTABLE[
V.DESCRIPTION[base.getMetaText(self.service, "description")],
V.RESOURCE(type="results")[
V.INFO(ID="Error", name="Error",
value=str(msg).replace('"', '\\"')),
V.INFO(name="request", value=utils.debytify(request.uri))]], request)
request.write("\n")
request.finish()
def _formatOutput(self, data, request, stream=True):
"""makes output SCS 1.02 compatible or causes the service to error out.
This comprises mapping meta.id;meta.main to ID_MAIN and
pos.eq* to POS_EQ*.
"""
if isinstance(data, tuple):
return DALRenderer._formatOutput(self, data, request, stream)
ucdCasts = {
"meta.id;meta.main": {"ucd": "ID_MAIN", "datatype": "char",
"arraysize": "*"},
"pos.eq.ra;meta.main": {"ucd": "POS_EQ_RA_MAIN",
"datatype": "double"},
"pos.eq.dec;meta.main": {"ucd": "POS_EQ_DEC_MAIN",
"datatype": "double"},
}
realCasts = {}
table = data.getPrimaryTable()
for ind, ofield in enumerate(table.tableDef.columns):
if ofield.ucd in ucdCasts:
realCasts[ofield.name] = ucdCasts.pop(ofield.ucd)
if ucdCasts:
return self._writeErrorTable(request, "Table cannot be formatted for"
" SCS. Column(s) with the following new UCD(s) were missing in"
" output table: %s"%', '.join(ucdCasts))
# allow integers as ID_MAIN [HACK -- this needs to become saner.
# conditional cast functions?]
idCol = table.tableDef.getColumnByUCD("meta.id;meta.main")
if idCol.type in set(["integer", "bigint", "smallint"]):
realCasts[idCol.name]["castFunction"] = str
table.votCasts = realCasts
return DALRenderer._formatOutput(self,
data, request, stream=stream)
[docs]class SIAPRenderer(DALRenderer):
"""A renderer for a the Simple Image Access Protocol.
These have errors in the content of an info element, and they support
metadata queries.
For registration, services using this renderer must set the following
metadata items:
- sia.type -- one of Cutout, Mosaic, Atlas, Pointed, see SIAP spec
You should set the following metadata items:
- testQuery.pos.ra, testQuery.pos.dec -- RA and Dec for a query that
yields at least one image
- testQuery.size.ra, testQuery.size.dec -- RoI extent for a query that
yields at least one image.
You can set the following metadata items (there are defaults on them
that basically communicate there are no reasonable limits on them):
- sia.maxQueryRegionSize.(long|lat)
- sia.maxImageExtent.(long|lat)
- sia.maxFileSize
- sia.maxRecord (default dalHardLimit global meta)
"""
version = "1.0"
name = "siap.xml"
parameterStyle = "pql"
standardId = "ivo://ivoa.net/std/sia"
_outputTableCasts = {
"pixelScale": {
"datatype": "double",
"arraysize": "*",
"ucd": "VOX:Image_Scale"},
"wcs_cdmatrix": {
"datatype": "double",
"arraysize": "*",
"ucd": "VOX:WCS_CDMatrix"},
"wcs_refValues": {
"datatype": "double",
"arraysize": "*",
"ucd": "VOX:WCS_CoordRefValue"},
"bandpassHi": {
"datatype": "double",
"ucd": "VOX:BandPass_HiLimit"},
"bandpassLo": {
"datatype": "double",
"ucd": "VOX:BandPass_LoLimit"},
"bandpassRefval": {
"datatype": "double",
"ucd": "VOX:BandPass_RefValue"},
"wcs_refPixel": {
"datatype": "double",
"arraysize": "*",
"ucd": "VOX:WCS_CoordRefPixel"},
"wcs_projection": {
"arraysize": "3",
"castFunction": lambda s: s[:3],
"ucd": "VOX:WCS_CoordProjection"},
"mime": {"ucd": "VOX:Image_Format"},
"accref": {"ucd": "VOX:Image_AccessReference"},
"accsize": {"datatype": "int"},
"centerAlpha": {"ucd": "POS_EQ_RA_MAIN"},
"centerDelta": {"ucd": "POS_EQ_DEC_MAIN"},
"imageTitle": {"ucd": "VOX:Image_Title"},
"instId": {"ucd": "INST_ID"},
"dateObs": {"ucd": "VOX:Image_MJDateObs"},
"nAxes": {"ucd": "VOX:Image_Naxes"},
"pixelSize": {"ucd": "VOX:Image_Naxis"},
"refFrame": {"ucd": "VOX:STC_CoordRefFrame"},
"wcs_equinox": {"ucd": "VOX:STC_CoordEquinox"},
"bandpassId": {"ucd": "VOX:BandPass_ID"},
"bandpassUnit": {"ucd": "VOX:BandPass_Unit"},
"pixflags": {"ucd": "VOX:Image_PixFlags"},
}
def _formatOutput(self, data, request, stream=True):
if hasattr(data, "setMeta"):
# let (media-type, payload) results pass by
data.setMeta("_type", "results")
data.getPrimaryTable().votCasts = self._outputTableCasts
return DALRenderer._formatOutput(self,
data, request,
stream=stream)
def _makeErrorTable(self, request, msg, queryStatus="ERROR"):
return V.VOTABLE[
V.RESOURCE(type="results")[
V.INFO(name="QUERY_STATUS", value=queryStatus)[
str(msg)]]]
[docs]class UnifiedDALRenderer(DALRenderer):
"""A renderer for new-style simple DAL protocols.
All input processing (e.g., metadata queries and the like) are considered
part of the individual protocol and thus left to the core.
The error style is that of SSAP (which, hopefully, will be kept
for the other DAL2 protocols, too).
To define actual renderers, inherit from this and set the name attribute
(plus _outputTableCasts if necessary). Also, explain any protocol-specific
metadata in the docstring.
"""
# TODO: merge this into DALRenderer and make legacy classes override
# whatever they need differently.
_outputTableCasts = {}
def _formatOutput(self, data, request, stream=True):
request.setHeader("content-type", "text/xml+votable")
if hasattr(data, "setMeta"):
data.setMeta("_type", "results")
data.getPrimaryTable().votCasts = self._outputTableCasts
return DALRenderer._formatOutput(self,
data, request, stream=stream)
def _makeErrorTable(self, request, msg, queryStatus="ERROR"):
return V.VOTABLE11[
V.RESOURCE(type="results")[
V.INFO(name="QUERY_STATUS", value=queryStatus)[
str(msg)]]]
[docs]class SIAP2Renderer(UnifiedDALRenderer):
"""A renderer for SIAPv2.
In general, if you want a SIAP2 service, you'll need something like the
obscore view in the underlying table.
"""
parameterStyle = "dali"
name = "siap2.xml"
standardId = "ivo://ivoa.net/std/sia"
def _makeErrorTable(self, request, msg, queryStatus="ERROR"):
# FatalFault, DefaultFault
return V.VOTABLE[
V.RESOURCE(type="results")[
V.INFO(name="QUERY_STATUS", value=queryStatus)[
str(msg)]]]
def _handleRandomFailure(self, failure, request):
base.ui.notifyFailure(failure)
return self._writeErrorTable(request,
"DefaultFault: "+failure.getErrorMessage(),
httpStatus=500)
def _handleInputErrors(self, failure, request):
httpStatus, queryStatus = 200, "ERROR"
if isinstance(failure.value, base.EmptyData):
httpStatus, queryStatus = 400, "EMPTY"
return self._writeErrorTable(request,
"UsageFault: "+failure.getErrorMessage(),
httpStatus=httpStatus,
queryStatus=queryStatus)
[docs]class SSAPRenderer(UnifiedDALRenderer):
"""A renderer for the simple spectral access protocol.
For registration, you must set the following metadata
for the ssap.xml renderer:
- ssap.dataSource -- survey, pointed, custom, theory, artificial
- ssap.testQuery -- a query string that returns some data; REQUEST=queryData
is added automatically
that describe the type of data served through the service. Will
usually by ``spectrum``, but ``timeseries`` is a realistic option.
Other SSA metadata includes:
- ssap.creationType -- archival, cutout, filtered, mosaic,
projection, spectralExtraction, catalogExtraction (defaults to archival)
- ssap.complianceLevel -- set to "query" when you don't deliver
SDM compliant spectra; otherwise don't say anything, DaCHS will fill
in the right value.
It is recommended to set this metadata globally on the RD, as the
SSA mixin can use that metadata to fill tables with sensible values
without operator intervention.
Properties supported by this renderer:
- datalink -- if present, this must be the id of a datalink service
that can work with the pubDIDs in this table (don't use this any more,
datalink is handled through table-level metadata now)
- defaultRequest -- by default, requests without a REQUEST parameter
will be rejected. If you set defaultRequest to querydata, such
requests will be processed as if REQUEST were given (which is of
course sane but is a violation of the standard).
"""
version = "1.04"
name = "ssap.xml"
parameterStyle = "pql"
standardId = "ivo://ivoa.net/std/ssap"
defaultOutputFormat = "votabletd"
def _getMetadataData(self):
data = UnifiedDALRenderer._getMetadataData(self)
data.dd.addMeta(
"info", "SSAP", infoName="SERVICE_PROTOCOL", infoValue="1.04")
return data
def _formatOutput(self, data, request, stream=True):
# for SSA, we need some funny attributes on the root resource
if hasattr(data, "setMeta"):
# the thing isn't already rendered, so we can add some additional
# metadata
data.contributingMetaCarriers.append(self.service)
data.setMeta("_type", "results")
data.addMeta("_votableRootAttributes",
'xmlns:ssa="http://www.ivoa.net/xml/DalSsap/v1.0"')
data.addMeta("info", "SSAP",
infoName="SERVICE_PROTOCOL",
infoValue=self.version)
# In SSA, we want a "direct" SODA block if we have a dlget-capable
# datalink service.
sodaGenerators = []
for table in data:
if not table.rows:
continue
for svcMeta in table.iterMeta("_associatedDatalinkService"):
dlService = base.resolveId(table.tableDef.rd,
base.getMetaText(svcMeta, "serviceId"))
if "dlget" not in dlService.allowed:
continue
if utils.looksLikeURLPat.match(table.rows[0]["accref"]):
# we need the product table for direct processing
# (if we ever want this, we'd have to discover the
# descriptor class from the datalink service)
continue
core = ssap.getDatalinkCore(dlService, table)
sodaGenerators.append(
lambda ctx, core=core, svcMeta=svcMeta, table=table:
core.datalinkEndpoints[0].asVOT(
ctx,
linkIdTo=ctx.getOrMakeIdFor(
table.tableDef.getByName(
base.getMetaText(svcMeta, "idColumn")))))
if sodaGenerators:
data.sodaGenerators = sodaGenerators
return UnifiedDALRenderer._formatOutput(self,
data, request, stream=stream)
[docs]class SLAPRenderer(UnifiedDALRenderer):
"""A renderer for the simple line access protocol SLAP.
For registration, you must set the following metadata on services
using the slap.xml renderer:
There's two mandatory metadata items for these:
- slap.dataSource -- one of observational/astrophysical,
observational/laboratory, or theoretical
- slap.testQuery -- parameters that lead to a non-empty response.
The way things are written in DaCHS, MAXREC=1 should in general
work.
"""
version = "1.0"
name = "slap.xml"
parameterStyle = "pql"
standardId = "ivo://ivoa.net/std/ssap"
def _formatOutput(self, data, request, stream=True):
if hasattr(data, "addMeta"):
# SLAP 1.0 requires QUERY_STATUS OK (rather than OVERFLOW). Oh my.
data.getPrimaryTable().setMeta("_queryStatus", "OK")
data.addMeta("_votableRootAttributes",
'xmlns:ssldm="http://www.ivoa.net/xml/SimpleSpectrumLineDM'
'/SimpleSpectrumLineDM-v1.0.xsd"')
return UnifiedDALRenderer._formatOutput(self,
data, request, stream=stream)
[docs]class APIRenderer(UnifiedDALRenderer):
"""A renderer that works like a VO standard renderer but that doesn't
actually follow a given protocol.
Use this for improvised APIs. The default output format is a VOTable,
and the errors come in VOSI VOTables. The renderer does, however,
evaluate basic DALI parameters. You can declare that by
including <FEED source="//pql#DALIPars"/> in your service.
These will return basic service metadata if passed MAXREC=0.
"""
name = "api"
parameterStyle = "dali"
[docs]class RegistryRenderer(grend.ServiceBasedPage):
"""A renderer that works with registry.oaiinter to provide an OAI-PMH
interface.
The core is expected to return a stanxml tree.
"""
name = "pubreg.xml"
urlUse = "base"
resultType = "text/xml"
[docs] def render(self, request):
# Make a robust (unchecked) pars dict for error rendering; real
# parameter checking happens in getPMHResponse
inData = {"args": request.strargs}
streamResponse = request.strargs.get("verb")==["ListRecords"]
self.runAsync(inData,
).addCallback(self._renderXML, request, streamResponse
).addErrback(self._renderError, request, inData["args"])
return server.NOT_DONE_YET
def _renderXML(self, resultTree, request, streamResponse=False):
request.setHeader("content-type", "text/xml")
def writeTo(f):
utils.xmlwrite(resultTree, f,
xmlDecl=True,
prolog="<?xml-stylesheet href='/static/xsl/oai.xsl' type='text/xsl'?>")
if streamResponse:
streaming.streamOut(writeTo, request, self.queryMeta)
else:
writeTo(request)
request.finish()
def _getErrorTree(self, exception, pars):
"""returns an ElementTree containing an OAI-PMH error response.
If exception is one of "our" exceptions, we translate them to error messages.
Otherwise, we reraise the exception to an enclosing
function may "handle" it.
Contrary to the recommendation in the OAI-PMH spec, this will only
return one error at a time.
"""
from gavo.registry.model import OAI
if isinstance(exception, registry.OAIError):
code = exception.__class__.__name__
code = code[0].lower()+code[1:]
message = str(exception)
else:
code = "badArgument" # Why the hell don't they have a serverError?
message = "Internal Error: "+str(exception)
return OAI.PMH[
OAI.responseDate[datetime.datetime.utcnow().strftime(
utils.isoTimestampFmt)],
OAI.request(metadataPrefix=pars.get("metadataPrefix", [None])[0]),
OAI.error(code=code)[
message
]
]
def _renderError(self, failure, request, pars):
if getattr(request, "channel", None) is None:
# remote has hung up; don't try to write to them.
return
try:
if not isinstance(failure.value,
(registry.OAIError, base.ValidationError)):
base.ui.notifyFailure(failure)
return self._renderXML(self._getErrorTree(failure.value, pars),
request)
except:
base.ui.notifyError("Cannot create registry error document")
request.setResponseCode(400)
request.setHeader("content-type", "text/plain")
request.write("Internal error. Please notify site maintainer")
request.finish()
class _DatalinkRendererBase(grend.ServiceBasedPage):
"""the base class of the two datalink sync renderers.
"""
urlUse = "base"
# send out files as attachments with separate file names?
attachResult = False
def render(self, request):
self.runAsync(request.strargs,
).addCallback(self._formatData, request
).addErrback(self._reportError, request
).addErrback(weberrors.renderDCErrorPage, request)
return server.NOT_DONE_YET
def _formatData(self, data, request):
# the core returns mime, data or a resource. So, if it's a pair,
# do something myself, else let twisted sort it out
if isinstance(data, tuple):
# XXX TODO: the same thing is in formrender. Refactor; since this is
# something most renderers should be able to do, ServiceBasedPage would be
# a good place
mime, payload = data
request.setHeader("content-type", mime)
if self.attachResult:
addAttachmentHeaders(request, mime)
return streaming.streamOut(
lambda f: f.write(payload), request, self.queryMeta)
else:
if self.attachResult:
# the following getattr is for when data is a nevow.static.File
addAttachmentHeaders(request, getattr(data, "type", None))
data.render(request)
failureNameMap = {
'ValidationError': 'UsageError',
'MultiplicityError': 'MultiValuedParamNotSupported',
}
def _reportError(self, failure, request):
# Do not trap svcs.WebRedirect here!
failure.trap(base.ValidationError, soda.EmptyData)
request.setHeader("content-type", "text/plain")
if hasattr(failure.value, "responseCode"):
request.setResponseCode(failure.value.responseCode)
else:
request.setResponseCode(422)
if hasattr(failure.value, "responsePayload"):
request.write(failure.value.responsePayload)
else:
request.write("%s: %s\n"%(
self.failureNameMap.get(failure.value.__class__.__name__, "Error"),
utils.safe_str(failure.value)))
request.finish()
return server.NOT_DONE_YET
def _doDatalinkXSLT(data, _cache={}):
"""a temporary hack to do server-side XSLT while the browser implementations
apparently suck.
Remove this once we've worked out how to make the datalink-to-xml
stylesheet compatible with actual browsers.
"""
if "etree" not in _cache:
from lxml import etree as lxmletree
_cache["etree"] = lxmletree
if "style" not in _cache:
with base.openDistFile("web/xsl/datalink-to-html.xsl", "rb") as f:
_cache["style"] = _cache["etree"].XSLT(
_cache["etree"].XML(f.read()))
return bytes(_cache["style"](_cache["etree"].XML(data)))
[docs]class DatalinkGetDataRenderer(_DatalinkRendererBase):
"""A renderer for data processing by datalink cores.
This must go together with a datalink core, nothing else will do.
This renderer will actually produce the processed data. It must be
complemented by the dlmeta renderer which allows retrieving metadata.
"""
name = "dlget"
attachResult = True
standardId = "ivo://ivoa.net/std/soda#sync-1.0"
# This shouldn't have parameterStyle for now, as it would
# add DALI parameters (MAXREC etc) which are probably inappropriate
# here.