Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _builder(cls):
return QueryBuilder(dialect=Dialects.REDSHIFT)
def _builder(cls):
return QueryBuilder(dialect=Dialects.SQLLITE, wrapper_cls=SQLLiteValueWrapper)
"""
@classmethod
def _builder(cls):
return VerticaQueryBuilder()
@classmethod
def from_file(cls, fp):
return VerticaCopyQueryBuilder().from_file(fp)
@classmethod
def create_table(cls, table):
return VerticaCreateQueryBuilder().create_table(table)
class OracleQueryBuilder(QueryBuilder):
def __init__(self):
super(OracleQueryBuilder, self).__init__(dialect=Dialects.ORACLE)
def get_sql(self, *args, **kwargs):
return super(OracleQueryBuilder, self).get_sql(
*args, groupby_alias=False, **kwargs
)
class OracleQuery(Query):
"""
Defines a query class for use with Oracle.
"""
@classmethod
def _builder(cls):
def translate_exceptions(func: F) -> F:
@wraps(func)
async def translate_exceptions_(self, *args):
try:
return await func(self, *args)
except pyodbc.ProgrammingError as exc:
raise OperationalError(exc)
except pyodbc.IntegrityError as exc:
raise IntegrityError(exc)
except pyodbc.Error as exc:
raise (_translate_pyodbc_err(exc))
return translate_exceptions_ # type: ignore
class OracleQueryBuilder(QueryBuilder): # type: ignore
def __init__(self, **kwargs):
super(OracleQueryBuilder, self).__init__(dialect=Dialects.ORACLE, **kwargs)
def get_sql(self, *args, **kwargs):
return super(OracleQueryBuilder, self).get_sql(*args, groupby_alias=False, **kwargs)
def _limit_sql(self):
return f" FETCH NEXT {self._limit} ROWS ONLY"
class OracleQuery(Query): # type: ignore
"""
Defines a query class for use with Oracle.
"""
@classmethod
FieldError,
IntegrityError,
MultipleObjectsReturned,
ParamsError,
)
from tortoise.expressions import F
from tortoise.fields.relational import (
ForeignKeyFieldInstance,
OneToOneFieldInstance,
RelationalField,
)
from tortoise.functions import Function
from tortoise.query_utils import Prefetch, Q, QueryModifier, _get_joins_for_related_field
# Empty placeholder - Should never be edited.
QUERY: QueryBuilder = QueryBuilder()
if TYPE_CHECKING: # pragma: nocoverage
from tortoise.models import Model
MODEL = TypeVar("MODEL", bound="Model")
T_co = TypeVar("T_co", covariant=True)
class QuerySetSingle(Protocol[T_co]):
"""
Awaiting on this will resolve a single instance of the Model object, and not a sequence.
"""
# pylint: disable=W0104
def __await__(self) -> Generator[Any, None, T_co]:
... # pragma: nocoverage
def _builder(cls):
return QueryBuilder(dialect=Dialects.CLICKHOUSE, wrap_union_queries=False)
return super(OracleQueryBuilder, self).get_sql(
*args, groupby_alias=False, **kwargs
)
class OracleQuery(Query):
"""
Defines a query class for use with Oracle.
"""
@classmethod
def _builder(cls):
return OracleQueryBuilder()
class PostgreQueryBuilder(QueryBuilder):
def __init__(self):
super(PostgreQueryBuilder, self).__init__(dialect=Dialects.POSTGRESQL)
self._returns = []
self._return_star = False
self._on_conflict_field = None
self._on_conflict_do_nothing = False
self._on_conflict_updates = []
def __copy__(self):
newone = super(PostgreQueryBuilder, self).__copy__()
newone._returns = copy(self._returns)
newone._on_conflict_updates = copy(self._on_conflict_updates)
return newone
@builder
def on_conflict(self, target_field):
def __init__(self):
super(SnowFlakeQueryBuilder, self).__init__(dialect=Dialects.SNOWFLAKE)
class SnowflakeQuery(Query):
"""
Defines a query class for use with Snowflake.
"""
@classmethod
def _builder(cls):
return SnowFlakeQueryBuilder()
class MySQLQueryBuilder(QueryBuilder):
QUOTE_CHAR = "`"
def __init__(self):
super(MySQLQueryBuilder, self).__init__(
dialect=Dialects.MYSQL, wrap_union_queries=False
)
self._duplicate_updates = []
self._modifiers = []
def __copy__(self):
newone = super(MySQLQueryBuilder, self).__copy__()
newone._duplicate_updates = copy(self._duplicate_updates)
return newone
@builder
def on_duplicate_key_update(self, field, value):
@classmethod
def _builder(cls):
return PostgreQueryBuilder()
class RedshiftQuery(Query):
"""
Defines a query class for use with Amazon Redshift.
"""
@classmethod
def _builder(cls):
return QueryBuilder(dialect=Dialects.REDSHIFT)
class MSSQLQueryBuilder(QueryBuilder):
def __init__(self):
super(MSSQLQueryBuilder, self).__init__(dialect=Dialects.MSSQL)
self._top = None
@builder
def top(self, value):
"""
Implements support for simple TOP clauses.
Does not include support for PERCENT or WITH TIES.
https://docs.microsoft.com/en-us/sql/t-sql/queries/top-transact-sql?view=sql-server-2017
"""
try:
self._top = int(value)
except ValueError:
class MySQLQuery(Query):
"""
Defines a query class for use with MySQL.
"""
@classmethod
def _builder(cls):
return MySQLQueryBuilder()
@classmethod
def load(cls, fp):
return MySQLLoadQueryBuilder().load(fp)
class VerticaQueryBuilder(QueryBuilder):
def __init__(self):
super(VerticaQueryBuilder, self).__init__(dialect=Dialects.VERTICA)
self._hint = None
@builder
def hint(self, label):
self._hint = label
def get_sql(self, *args, **kwargs):
sql = super(VerticaQueryBuilder, self).get_sql(*args, **kwargs)
if self._hint is not None:
sql = "".join(
[sql[:7], "/*+label({hint})*/".format(hint=self._hint), sql[6:]]
)