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)