# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors # License: MIT. See LICENSE import typing from functools import cached_property from types import NoneType import frappe from frappe.query_builder.builder import MariaDB, Postgres from frappe.query_builder.functions import Function if typing.TYPE_CHECKING: from frappe.query_builder import DocType Query = str | MariaDB | Postgres QueryValues = tuple | list | dict | NoneType EmptyQueryValues = object() FallBackDateTimeStr = "0001-01-01 00:00:00.000000" NestedSetHierarchy = ( "ancestors of", "descendants of", "not ancestors of", "not descendants of", ) def is_query_type(query: str, query_type: str | tuple[str]) -> bool: return query.lstrip().split(maxsplit=1)[0].lower().startswith(query_type) def is_pypika_function_object(field: str) -> bool: return getattr(field, "__module__", None) == "pypika.functions" or isinstance(field, Function) def get_doctype_name(table_name: str) -> str: if table_name.startswith(("tab", "`tab", '"tab')): table_name = table_name.replace("tab", "", 1) table_name = table_name.replace("`", "") table_name = table_name.replace('"', "") return table_name class LazyString: def _setup(self) -> None: raise NotImplementedError @cached_property def value(self) -> str: return self._setup() def __str__(self) -> str: return self.value def __repr__(self) -> str: return f"'{self.value}'" class LazyDecode(LazyString): __slots__ = () def __init__(self, value: str) -> None: self._value = value def _setup(self) -> None: return self._value.decode() class LazyMogrify(LazyString): __slots__ = () def __init__(self, query, values) -> None: self.query = query self.values = values def _setup(self) -> str: return frappe.db.mogrify(self.query, self.values)