70 lines
1.5 KiB
Python
70 lines
1.5 KiB
Python
# 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)
|
|
|
|
|
|
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)
|