128 lines
4 KiB
Python
128 lines
4 KiB
Python
import re
|
|
import types
|
|
import typing
|
|
|
|
from pypika import MySQLQuery, Order, PostgreSQLQuery, SQLLiteQuery, terms
|
|
from pypika.dialects import MySQLQueryBuilder, PostgreSQLQueryBuilder, SQLLiteQueryBuilder
|
|
from pypika.queries import QueryBuilder, Schema, Table
|
|
from pypika.terms import Function
|
|
|
|
from frappe.query_builder.terms import ParameterizedValueWrapper, SQLiteParameterizedValueWrapper
|
|
from frappe.utils import get_table_name
|
|
|
|
# less restrictive version of frappe.core.doctype.doctype.doctype.START_WITH_LETTERS_PATTERN
|
|
# to allow table names like __Auth
|
|
TABLE_NAME_PATTERN = re.compile(r"^[\w -]*$", flags=re.ASCII)
|
|
|
|
|
|
class Base:
|
|
terms = terms
|
|
desc = Order.desc
|
|
asc = Order.asc
|
|
Schema = Schema
|
|
Table = Table
|
|
|
|
# Added dynamic type hints for engine attribute
|
|
# which is to be assigned later.
|
|
if typing.TYPE_CHECKING:
|
|
from frappe.database.query import Engine
|
|
|
|
engine: Engine
|
|
|
|
@staticmethod
|
|
def functions(name: str, *args, **kwargs) -> Function:
|
|
return Function(name, *args, **kwargs)
|
|
|
|
@staticmethod
|
|
def DocType(table_name: str, *args, **kwargs) -> Table:
|
|
Base.validate_doctype(table_name)
|
|
table_name = get_table_name(table_name)
|
|
return Table(table_name, *args, **kwargs)
|
|
|
|
@classmethod
|
|
def into(cls, table, *args, **kwargs) -> QueryBuilder:
|
|
if isinstance(table, str):
|
|
table = cls.DocType(table)
|
|
return super().into(table, *args, **kwargs)
|
|
|
|
@classmethod
|
|
def update(cls, table, *args, **kwargs) -> QueryBuilder:
|
|
if isinstance(table, str):
|
|
table = cls.DocType(table)
|
|
return super().update(table, *args, **kwargs)
|
|
|
|
@staticmethod
|
|
def validate_doctype(doctype) -> None:
|
|
from frappe import _, throw
|
|
|
|
if not TABLE_NAME_PATTERN.match(doctype):
|
|
throw(_("Invalid DocType: {0}").format(doctype))
|
|
|
|
|
|
class MariaDB(Base, MySQLQuery):
|
|
Field = terms.Field
|
|
|
|
_BuilderClasss = MySQLQueryBuilder
|
|
|
|
@classmethod
|
|
def _builder(cls, *args, **kwargs) -> "MySQLQueryBuilder":
|
|
return super()._builder(*args, wrapper_cls=ParameterizedValueWrapper, **kwargs)
|
|
|
|
@classmethod
|
|
def from_(cls, table, *args, **kwargs):
|
|
if isinstance(table, str):
|
|
table = cls.DocType(table)
|
|
return super().from_(table, *args, **kwargs)
|
|
|
|
|
|
class Postgres(Base, PostgreSQLQuery):
|
|
field_translation = types.MappingProxyType({"table_name": "relname", "table_rows": "n_tup_ins"})
|
|
schema_translation = types.MappingProxyType({"tables": "pg_stat_all_tables"})
|
|
# TODO: Find a better way to do this
|
|
# These are interdependent query changes that need fixing. These
|
|
# translations happen in the same query. But there is no check to see if
|
|
# the Fields are changed only when a particular `information_schema` schema
|
|
# is used. Replacing them is not straightforward because the "from_"
|
|
# function can not see the arguments passed to the "select" function as
|
|
# they are two different objects. The quick fix used here is to replace the
|
|
# Field names in the "Field" function.
|
|
|
|
_BuilderClasss = PostgreSQLQueryBuilder
|
|
|
|
@classmethod
|
|
def _builder(cls, *args, **kwargs) -> "PostgreSQLQueryBuilder":
|
|
return super()._builder(*args, wrapper_cls=ParameterizedValueWrapper, **kwargs)
|
|
|
|
@classmethod
|
|
def Field(cls, field_name, *args, **kwargs):
|
|
if field_name in cls.field_translation:
|
|
field_name = cls.field_translation[field_name]
|
|
return terms.Field(field_name, *args, **kwargs)
|
|
|
|
@classmethod
|
|
def from_(cls, table, *args, **kwargs):
|
|
if isinstance(table, Table):
|
|
if table._schema:
|
|
if table._schema._name == "information_schema":
|
|
table = cls.schema_translation.get(table._table_name) or table
|
|
|
|
elif isinstance(table, str):
|
|
table = cls.DocType(table)
|
|
|
|
return super().from_(table, *args, **kwargs)
|
|
|
|
|
|
class SQLite(Base, SQLLiteQuery):
|
|
Field = terms.Field
|
|
|
|
_BuilderClasss = SQLLiteQueryBuilder
|
|
|
|
@classmethod
|
|
def _builder(cls, *args, **kwargs) -> "SQLLiteQueryBuilder":
|
|
return super()._builder(*args, wrapper_cls=SQLiteParameterizedValueWrapper, **kwargs)
|
|
|
|
@classmethod
|
|
def from_(cls, table, *args, **kwargs):
|
|
if isinstance(table, str):
|
|
table = cls.DocType(table)
|
|
return super().from_(table, *args, **kwargs)
|