seitime-frappe/frappe/database/utils.py
Akhil Narang e41fa2dfc7
feat: migrate columns to be non-nullable if required
Signed-off-by: Akhil Narang <me@akhilnarang.dev>
2023-11-16 14:51:57 +05:30

74 lines
1.7 KiB
Python

# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
# License: MIT. See LICENSE
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
Query = str | MariaDB | Postgres
QueryValues = tuple | list | dict | NoneType
EmptyQueryValues = object()
FallBackDateTimeStr = "0001-01-01 00:00:00.000000"
DefaultOrderBy = "KEEP_DEFAULT_ORDERING"
NestedSetHierarchy = (
"ancestors of",
"descendants of",
"not ancestors of",
"not descendants of",
"descendants of (inclusive)",
)
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("`", "")
return table_name.replace('"', "")
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)