Python sqlalchemy.sql 模块，literal_column() 实例源码

def literal_column(text, type_=None):
    """Return a textual column expression, as would be in the columns
    clause of a ``SELECT`` statement.

    The object returned supports further expressions in the same way as any
    other column object, including comparison, math and string operations.
    The type\_ parameter is important to determine proper expression behavior
    (such as, '+' means string concatenation or numerical addition based on
    the type).

    :param text: the text of the expression; can be any SQL expression.
      Quoting rules will not be applied. To specify a column-name expression
      which should be subject to quoting rules, use the :func:`column`
      function.

    :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
      object which will
      provide result-set translation and additional expression semantics for
      this column. If left as None the type will be NullType.

    """
    return ColumnClause(text, type_=type_, is_literal=True)

def column(text, type_=None):
    """Return a textual column clause, as would be in the columns clause of a
    ``SELECT`` statement.

    The object returned is an instance of :class:`.ColumnClause`, which
    represents the "syntactical" portion of the schema-level
    :class:`~sqlalchemy.schema.Column` object.  It is often used directly
    within :func:`~.expression.select` constructs or with lightweight
    :func:`~.expression.table` constructs.

    Note that the :func:`~.expression.column` function is not part of
    the ``sqlalchemy`` namespace.  It must be imported from the
    ``sql`` package::

        from sqlalchemy.sql import table, column

    :param text: the name of the column.  Quoting rules will be applied
      to the clause like any other column name. For textual column constructs
      that are not to be quoted, use the :func:`literal_column` function.

    :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` object
      which will provide result-set translation for this column.

    See :class:`.ColumnClause` for further examples.

    """
    return ColumnClause(text, type_=type_)

def _interpret_as_column_or_from(element):
    if isinstance(element, Visitable):
        return element
    elif hasattr(element, '__clause_element__'):
        return element.__clause_element__()

    insp = inspection.inspect(element, raiseerr=False)
    if insp is None:
        if isinstance(element, (util.NoneType, bool)):
            return _const_expr(element)
    elif hasattr(insp, "selectable"):
        return insp.selectable

    return literal_column(str(element))

def __init__(self, *args, **kwargs):
        if args and isinstance(args[0], (SelectBase, ScalarSelect)):
            s = args[0]
        else:
            if not args:
                args = ([literal_column('*')],)
            s = select(*args, **kwargs).as_scalar().self_group()

        UnaryExpression.__init__(self, s, operator=operators.exists,
                                  type_=sqltypes.Boolean)

def query(table):
    col1 = literal_column("TIMESTAMP_TRUNC(timestamp, DAY)").label("timestamp_label")
    col2 = func.sum(table.c.integer)
    query = (
        select([
            col1,
            col2,
        ])
        .where(col1 < '2017-01-01 00:00:00')
        .group_by(col1)
        .order_by(col2)
    )
    return query

def visit_BYTEINT(self, type_, **kw):
       return 'BYTEINT'



#@compiles(Select, 'teradata')
#def compile_select(element, compiler, **kw):
#    """
#    """
#
#    if not getattr(element, '_window_visit', None):
#      if element._limit is not None or element._offset is not None:
#          limit, offset = element._limit, element._offset
#
#          orderby=compiler.process(element._order_by_clause)
#          if orderby:
#            element = element._generate()
#            element._window_visit=True
#            #element._limit = None
#            #element._offset = None  cant set to none...
#
#            # add a ROW NUMBER() OVER(ORDER BY) column
#            element = element.column(sql.literal_column('ROW NUMBER() OVER (ORDER BY %s)' % orderby).label('rownum')).order_by(None)
#
#            # wrap into a subquery
#            limitselect = sql.select([c for c in element.alias().c if c.key != 'rownum'])
#
#            limitselect._window_visit=True
#            limitselect._is_wrapper=True
#
#            if offset is not None:
#              limitselect.append_whereclause(sql.column('rownum') > offset)
#              if limit is not None:
#                  limitselect.append_whereclause(sql.column('rownum') <= (limit + offset))
#            else:
#              limitselect.append_whereclause(sql.column("rownum") <= limit)
#
#            element = limitselect
#
#    kw['iswrapper'] = getattr(element, '_is_wrapper', False)
#    return compiler.visit_select(element, **kw)

def type_fields(schema, row):
    missing_values = []
    if 'missingValues' in schema._Schema__descriptor:
        missing_values = schema._Schema__descriptor['missingValues']

    typed_row = []
    for index, field in enumerate(schema.fields):
        value = row[index]
        if field.type == 'geojson':
            if value == '' or value == 'NULL' or value == None:
                value = None
            else:
                value = literal_column("ST_GeomFromGeoJSON('{}')".format(value))
        elif field.type == 'string' and 'None' not in missing_values and value == 'None':
            value = 'None'
        elif field.type == 'string' and value.lower() == 'nan':
            value = value # HACK: tableschema-py 1.0 fixes this but is not released yet
        elif field.type == 'array' or field.type == 'object':
            if value in missing_values:
                value = None
            else:
                value = literal_column('\'' + value + '\'::jsonb')
        else:
            try:
                value = field.cast_value(value)
            except InvalidObjectType:
                value = json.loads(value)

        if isinstance(value, datetime):
            value = literal_column("'" + value.strftime('%Y-%m-%d %H:%M:%S') + "'")
        elif isinstance(value, date):
            value = literal_column("'" + value.strftime('%Y-%m-%d') + "'")

        if value is None:
            value = literal_column('null')

        typed_row.append(value)

    return typed_row

def case(whens, value=None, else_=None):
    """Produce a ``CASE`` statement.

    whens
      A sequence of pairs, or alternatively a dict,
      to be translated into "WHEN / THEN" clauses.

    value
      Optional for simple case statements, produces
      a column expression as in "CASE <expr> WHEN ..."

    else\_
      Optional as well, for case defaults produces
      the "ELSE" portion of the "CASE" statement.

    The expressions used for THEN and ELSE,
    when specified as strings, will be interpreted
    as bound values. To specify textual SQL expressions
    for these, use the :func:`literal_column`
    construct.

    The expressions used for the WHEN criterion
    may only be literal strings when "value" is
    present, i.e. CASE table.somecol WHEN "x" THEN "y".
    Otherwise, literal strings are not accepted
    in this position, and either the text(<string>)
    or literal(<string>) constructs must be used to
    interpret raw string values.

    Usage examples::

      case([(orderline.c.qty > 100, item.c.specialprice),
            (orderline.c.qty > 10, item.c.bulkprice)
          ], else_=item.c.regularprice)
      case(value=emp.c.type, whens={
              'engineer': emp.c.salary * 1.1,
              'manager':  emp.c.salary * 3,
          })

    Using :func:`literal_column()`, to allow for databases that
    do not support bind parameters in the ``then`` clause.  The type
    can be specified which determines the type of the :func:`case()` construct
    overall::

        case([(orderline.c.qty > 100,
                literal_column("'greaterthan100'", String)),
              (orderline.c.qty > 10, literal_column("'greaterthan10'",
                String))
            ], else_=literal_column("'lethan10'", String))

    """

    return Case(whens, value=value, else_=else_)