/usr/lib/python3/dist-packages/sqlalchemy_utils/expressions.py is in python3-sqlalchemy-utils 0.32.21-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import (
_literal_as_text,
ClauseElement,
ColumnElement,
Executable,
FunctionElement
)
from sqlalchemy.sql.functions import GenericFunction
from .functions.orm import quote
class explain(Executable, ClauseElement):
"""
Define EXPLAIN element.
http://www.postgresql.org/docs/devel/static/sql-explain.html
"""
def __init__(
self,
stmt,
analyze=False,
verbose=False,
costs=True,
buffers=False,
timing=True,
format='text'
):
self.statement = _literal_as_text(stmt)
self.analyze = analyze
self.verbose = verbose
self.costs = costs
self.buffers = buffers
self.timing = timing
self.format = format
class explain_analyze(explain):
def __init__(self, stmt, **kwargs):
super(explain_analyze, self).__init__(
stmt,
analyze=True,
**kwargs
)
@compiles(explain, 'postgresql')
def pg_explain(element, compiler, **kw):
text = "EXPLAIN "
options = []
if element.analyze:
options.append('ANALYZE true')
if not element.timing:
options.append('TIMING false')
if element.buffers:
options.append('BUFFERS true')
if element.format != 'text':
options.append('FORMAT %s' % element.format)
if element.verbose:
options.append('VERBOSE true')
if not element.costs:
options.append('COSTS false')
if options:
text += '(%s) ' % ', '.join(options)
text += compiler.process(element.statement)
return text
class array_get(FunctionElement):
name = 'array_get'
@compiles(array_get)
def compile_array_get(element, compiler, **kw):
args = list(element.clauses)
if len(args) != 2:
raise Exception(
"Function 'array_get' expects two arguments (%d given)." %
len(args)
)
if not hasattr(args[1], 'value') or not isinstance(args[1].value, int):
raise Exception(
"Second argument should be an integer."
)
return '(%s)[%s]' % (
compiler.process(args[0]),
sa.text(str(args[1].value + 1))
)
class row_to_json(GenericFunction):
name = 'row_to_json'
type = postgresql.JSON
@compiles(row_to_json, 'postgresql')
def compile_row_to_json(element, compiler, **kw):
return "%s(%s)" % (element.name, compiler.process(element.clauses))
class json_array_length(GenericFunction):
name = 'json_array_length'
type = sa.Integer
@compiles(json_array_length, 'postgresql')
def compile_json_array_length(element, compiler, **kw):
return "%s(%s)" % (element.name, compiler.process(element.clauses))
class array_agg(GenericFunction):
name = 'array_agg'
type = postgresql.ARRAY
def __init__(self, arg, default=None, **kw):
self.type = postgresql.ARRAY(arg.type)
self.default = default
GenericFunction.__init__(self, arg, **kw)
@compiles(array_agg, 'postgresql')
def compile_array_agg(element, compiler, **kw):
compiled = "%s(%s)" % (element.name, compiler.process(element.clauses))
if element.default is None:
return compiled
return str(sa.func.coalesce(
sa.text(compiled),
sa.cast(postgresql.array(element.default), element.type)
).compile(compiler))
class Asterisk(ColumnElement):
def __init__(self, selectable):
self.selectable = selectable
@compiles(Asterisk)
def compile_asterisk(element, compiler, **kw):
return '%s.*' % quote(compiler.dialect, element.selectable.name)
|