Source code for gavo.svcs.outputdef

"""
Output tables and their components.
"""

#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 fnmatch

from gavo import base
from gavo import rscdef
from gavo import utils

_EMPTY_TABLE = base.makeStruct(rscdef.TableDef, id="<builtin empty table>")
_EMPTY_TABLE.getFullId = lambda: None


[docs]class OutputField(rscdef.Column): """A column for defining the output of a service. It adds some attributes useful for rendering results, plus functionality specific to certain cores. The optional formatter overrides the standard formatting code in HTML (which is based on units, ucds, and displayHints). You receive the item from the database as data and must return a string or t.w.template stan. In addition to the standard `Functions available for row makers`_ you have queryMeta and t.w.template's tags in T. Here's an example for generating a link to another service using this facility:: <outputField name="more" select="array[centerAlpha,centerDelta]" tablehead="More" description="More exposures near the center of this plate"> <formatter><![CDATA[ return T.a(href=base.makeSitePath("/lswscans/res/positions/q/form?" "POS=%s,%s&SIZE=1&INTERSECT=OVERLAPS&cutoutSize=0.5" "&__nevow_form__=genForm"%tuple(data) ))["More"] ]]> </formatter> </outputField> Within the code, in addition to data, you see rd and queryMeta. """ name_ = "outputField" _formatter = base.UnicodeAttribute("formatter", description="Function" " body to render this item to HTML.", copyable=True, expand=True) _wantsRow = base.BooleanAttribute("wantsRow", description="Does" " formatter expect the entire row rather than the column value only?", copyable="True") _select = base.UnicodeAttribute("select", description="Use this SQL" " fragment rather than field name in the select list of a DB based" " core.", default=base.Undefined, copyable=True, expand=True) _sets = base.StringSetAttribute("sets", description= "Output sets this field should be included in; ALL includes the field" " in all output sets.", copyable=True) def __repr__(self): return "<OutputField %s>"%repr(self.name)
[docs] def completeElement(self, ctx): if self.restrictedMode and ( self.formatter or self.select): raise base.RestrictedElement(self.name_, hint="formatter and select" " attributes on output fields are not allowed in restricted mode.") if self.select is base.Undefined: self.select = self.name super().completeElement(ctx)
[docs] @classmethod def fromColumn(cls, col): res = cls(None, **col.getAttributes(rscdef.Column)) res.stc = col.stc res.dmRoles = rscdef.OldRoles(col.dmRoles) return res.finishElement()
[docs] def expand(self, *args, **kwargs): return self.parent.expand(*args, **kwargs)
[docs]class OutputTableDef(rscdef.TableDef): """A table that has outputFields for columns. Cores always have one of these, but they are implicitly defined by the underlying database tables in case of dbCores and such. Services may define output tables to modify what is coming back fromt the core. Note that this usually only affects the output to web browsers. To use the output table also through VO protocols (and when producing VOTables, FITS files, and the like), you need to set the service's votableRespectsOutputTable property to True. """ name_ = "outputTable" # Don't validate meta for these -- while they are children # of validated structures (services), they don't need any # meta at all. This should go as soon as we have a sane # inheritance hierarchy for tables. metaModel = None _cols = rscdef.ColumnListAttribute("columns", childFactory=OutputField, description="Output fields for this table.", aliases=["column"], copyable=True) _verbLevel = base.IntAttribute("verbLevel", default=None, description="Copy over columns from fromTable not" " more verbose than this.") _autocols = base.StringListAttribute("autoCols", description="Column names obtained from fromTable; you can use" " shell patterns into the output table's parent table (in a table" " core, that's the queried table; in a service, it's the core's" " output table) here.") def __init__(self, parent, **kwargs): rscdef.TableDef.__init__(self, parent, **kwargs) self.parentTable = None try: # am I in a table-based core? self.parentTable = self.parent.queriedTable except (AttributeError, base.StructureError): # no. pass if not self.parentTable: try: # am I in a service with a core with output table? self.parentTable = self.parent.core.outputTable except (AttributeError, base.StructureError): # no. pass if not self.parentTable: # no suitable column source, use an empty table: self.parentTable = _EMPTY_TABLE self.namePath = None def _adoptColumn(self, sourceColumn): # Do not overwrite existing fields here to let the user # override individually try: self.getColumnByName(sourceColumn.name) except base.NotFoundError: self.feedObject("outputField", OutputField.fromColumn(sourceColumn)) def _addNames(self, ctx, names): # since autoCols is not copyable, we can require # that _addNames only be called when there's a real parse context. if ctx is None: raise base.StructureError("outputTable autocols is" " only available with a parse context") for name in names: self._addName(ctx, name) def _addName(self, ctx, name): """adopts a param or column name into the outputTable. name may be a reference or a param or column name in the parent table (as determined in the constructor, i.e., the queried table of a core or the output table of a service's core. You can also use shell patterns into parent columns. """ if utils.identifierPattern.match(name): refOb = ctx.resolveId(name, self) if refOb.name_=="param": self.feedObject("param", refOb.copy(self)) else: self._adoptColumn(refOb) else: # it's a shell pattern into parent table for col in self.parentTable: if fnmatch.fnmatch(col.name, name): self._adoptColumn(col)
[docs] def completeElement(self, ctx): if self.autoCols: self._addNames(ctx, self.autoCols) if self.verbLevel: table = self.parentTable for col in table.columns: if col.verbLevel<=self.verbLevel: self._adoptColumn(col) for par in table.params: if par.verbLevel<=self.verbLevel: self.feedObject("param", par.copy(self)) super().completeElement(ctx)
[docs] @classmethod def fromColumns(cls, columns, **kwargs): return rscdef.TableDef.fromColumns([OutputField.fromColumn(c) for c in columns])
[docs] @classmethod def fromTableDef(cls, tableDef, ctx=None): res = cls(None, columns=[OutputField.fromColumn(c) for c in tableDef], dupePolicy=tableDef.dupePolicy, primary=tableDef.primary, params=tableDef.params).finishElement(ctx) res.copyMetaFrom(tableDef) return res
# TODO: we ought to somewhere have an API like # # def getTableForQuery(td, colNames, whereClause, conn): # oTD = svcs.OutputTable.fromColumns( # [td.getColumnByName(n) for n in colNames]) # return rsc.TableForDef(ot, rows # =conn.queryToDicts(td.getSimpleQuery(oTD, whereClause))