Gmail warns you before posting something with "I've attached" and no
attachment, but apparently Google Groups does not.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/groups/opt_out.
from sqlalchemy.sql.expression import FromClause, ColumnClause, _anonymous_label, _truncated_label, func
from sqlalchemy.ext.compiler import compiles
__all__ = ['PGSetReturningFunction', 'srfunc', 'compiles_srf']
class PGSetReturningFunction(FromClause):
named_with_column = True
def __init__(self, func, columns, alias=None):
self.func = func
self._column_args = [ ColumnClause(c) if isinstance(c, str) else c for c in columns ]
self.name = alias or func.name
self._from_objects = [self]
def alias(self, name=None):
if name is None:
if self.named_with_column:
name = getattr(self, 'name', None)
name = _anonymous_label('%%(%d %s)s' % (id(self), name or 'anon'))
return self.__class__(self.func, self.c, name)
def column_alias(self, from_, to):
pg = self.__class__(self.func, self.c, self.name)
pg.c
pg._columns[to] = pg._columns[from_]._make_proxy(pg, to)
pg._columns.remove(pg.c[from_])
return pg
def column_map(self, iterable):
pg = self.__class__(self.func, self.c, self.name)
pg.c
_new_columns = dict()
for from_, to in iterable:
_new_columns[to] = pg._columns[from_]._make_proxy(pg, to)
pg._columns.remove(pg.c[from_])
pg._columns.update(_new_columns)
return pg
def _populate_column_collection(self):
for c in self._column_args:
c._make_proxy(self)
class Generator():
def __getattr__(self, key):
def create_function(*args, **kw):
columns = kw['columns']
return PGSetReturningFunction(getattr(func, key)(*args), columns)
return create_function
srfunc = PGSetReturningFunction.Generator()
@compiles(PGSetReturningFunction)
def compile_srf(element, compiler, **kw):
col_kwargs = kw.copy()
col_kwargs['include_table'] = False
alias=element.name
if isinstance(alias, _truncated_label):
alias = compiler._truncated_identifier("alias", alias)
return "{function} AS {alias}({columns})".format(
function=compiler.process(element.func, **kw),
alias=alias,
columns=", ".join(compiler.process(expr, **col_kwargs) for expr in element.c)
)
series = srfunc.generate_series(1, 10, columns=['value']).alias().column_map({'value': 'v'}.items())
print(str(series))
