Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
order = ' ORDER BY %s' % orderby
if limitby:
(lmin, lmax) = limitby
if lmin == 0:
dst += ' TOP %i' % lmax
else:
return self._rebuild_select_for_limit(
fields, tables, dst, whr, grp, order, lmin, lmax
)
if for_update:
upd = ' FOR UPDATE'
return 'SELECT%s %s FROM %s%s%s%s%s%s%s;' % (
dst, fields, tables, whr, grp, order, limit, offset, upd)
@dialects.register_for(MSSQL4)
class MSSQL4Dialect(MSSQL3Dialect):
def select(self, fields, tables, where=None, groupby=None, having=None,
orderby=None, limitby=None, distinct=False, for_update=False):
dst, whr, grp, order, limit, offset, upd = '', '', '', '', '', '', ''
if distinct is True:
dst = ' DISTINCT'
elif distinct:
dst = ' DISTINCT ON (%s)' % distinct
if where:
whr = ' %s' % self.where(where)
if groupby:
grp = ' GROUP BY %s' % groupby
if having:
grp += ' HAVING %s' % having
if orderby:
order = ' ORDER BY %s' % orderby
from .._compat import basestring
from ..adapters.mssql import MSSQL, MSSQLN, MSSQL3, MSSQL4, MSSQL3N, MSSQL4N, \
Vertica, Sybase
from ..helpers.methods import varquote_aux
from ..objects import Expression
from .base import SQLDialect
from . import dialects, sqltype_for
@dialects.register_for(MSSQL)
class MSSQLDialect(SQLDialect):
true = 1
false = 0
true_exp = '1=1'
false_exp = '1=0'
dt_sep = "T"
@sqltype_for('boolean')
def type_boolean(self):
return 'BIT'
@sqltype_for('blob')
def type_blob(self):
return 'IMAGE'
@sqltype_for('integer')
from .._gae import ndb
from ..adapters.google import GoogleDatastore
from ..helpers.gae import NDBDecimalProperty
from .base import NoSQLDialect
from . import dialects, sqltype_for
@dialects.register_for(GoogleDatastore)
class GoogleDatastoreDialect(NoSQLDialect):
FILTER_OPTIONS = {
"=": lambda a, b: a == b,
">": lambda a, b: a > b,
"<": lambda a, b: a < b,
"<=": lambda a, b: a <= b,
">=": lambda a, b: a >= b,
"!=": lambda a, b: a != b,
"in": lambda a, b: a.IN(b),
}
@sqltype_for("string")
def type_string(self):
return lambda **kwargs: ndb.StringProperty(**kwargs)
@sqltype_for("boolean")
if for_update:
upd = " FOR UPDATE"
return "SELECT%s%s%s %s FROM %s%s%s%s%s;" % (
dst,
offset,
limit,
fields,
tables,
whr,
grp,
order,
upd,
)
@dialects.register_for(InformixSE)
class InformixSEDialect(InformixDialect):
def select(
self,
fields,
tables,
where=None,
groupby=None,
having=None,
orderby=None,
limitby=None,
distinct=False,
for_update=False,
):
dst, whr, grp, order, limit, offset, upd = "", "", "", "", "", "", ""
if distinct is True:
dst = " DISTINCT"
from ..adapters.informix import Informix, InformixSE
from .firebird import FireBirdDialect
from . import dialects, sqltype_for
@dialects.register_for(Informix)
class InformixDialect(FireBirdDialect):
@sqltype_for("id")
def type_id(self):
return "SERIAL"
@sqltype_for("big-id")
def type_big_id(self):
return "BIGSERIAL"
@sqltype_for("reference FK")
def type_reference_fk(self):
return (
"REFERENCES %(foreign_key)s ON DELETE %(on_delete_action)s "
+ "CONSTRAINT FK_%(table_name)s_%(field_name)s"
)
from ..adapters.firebird import FireBird
from ..objects import Expression
from .base import SQLDialect
from . import dialects, sqltype_for
@dialects.register_for(FireBird)
class FireBirdDialect(SQLDialect):
@sqltype_for("text")
def type_text(self):
return "BLOB SUB_TYPE 1"
@sqltype_for("bigint")
def type_bigint(self):
return "BIGINT"
@sqltype_for("double")
def type_double(self):
return "DOUBLE PRECISION"
@sqltype_for("decimal")
def type_decimal(self):
return "DECIMAL(%(precision)s,%(scale)s)"
from ..adapters.postgres import Postgre, PostgreNew, PostgreBoolean
from ..helpers.methods import varquote_aux
from ..objects import Expression
from .base import SQLDialect
from . import dialects, sqltype_for, register_expression
@dialects.register_for(Postgre)
class PostgreDialect(SQLDialect):
true_exp = "TRUE"
false_exp = "FALSE"
@sqltype_for('blob')
def type_blob(self):
return 'BYTEA'
@sqltype_for('bigint')
def type_bigint(self):
return 'BIGINT'
@sqltype_for('double')
def type_double(self):
return 'FLOAT8'
from .._compat import basestring
from ..adapters.teradata import Teradata
from .base import SQLDialect
from . import dialects, sqltype_for
@dialects.register_for(Teradata)
class TeradataDialect(SQLDialect):
@sqltype_for("integer")
def type_integer(self):
return "INT"
@sqltype_for("text")
def type_text(self):
return "VARCHAR(2000)"
@sqltype_for("json")
def type_json(self):
return "VARCHAR(4000)"
@sqltype_for("float")
def type_float(self):
return "REAL"
return self.quote_template % val
def varquote(self, val):
return val
def sequence_name(self, tablename):
return self.quote("%s_sequence" % tablename)
def trigger_name(self, tablename):
return "%s_sequence" % tablename
def coalesce_zero(self, val, query_env={}):
return self.coalesce(val, [0], query_env)
@dialects.register_for(SQLAdapter)
class SQLDialect(CommonDialect):
quote_template = '"%s"'
true = "T"
false = "F"
true_exp = "1"
false_exp = "0"
dt_sep = " "
@sqltype_for("string")
def type_string(self):
return "VARCHAR(%(length)s)"
@sqltype_for("boolean")
def type_boolean(self):
return "CHAR(1)"
@register_expression('millenium')
def extract_millenium(self, expr):
return Expression(expr.db, self.extract, expr, 'millenium', 'integer')
class PostgreDialectJSON(PostgreDialect):
@sqltype_for('json')
def type_json(self):
return 'JSON'
@sqltype_for('jsonb')
def type_jsonb(self):
return 'JSONB'
@dialects.register_for(PostgreNew)
class PostgreDialectArrays(PostgreDialect):
@sqltype_for('list:integer')
def type_list_integer(self):
return 'BIGINT[]'
@sqltype_for('list:string')
def type_list_string(self):
return 'TEXT[]'
@sqltype_for('list:reference')
def type_list_reference(self):
return 'BIGINT[]'
def any(self, val, query_env={}):
return "ANY(%s)" % self.expand(val, query_env=query_env)