airflow_scheduler/.venv/Lib/site-packages/flask_appbuilder/models/base.py

import datetime
from functools import reduce
import logging
from typing import Any, Type

from flask_babel import lazy_gettext

from .filters import BaseFilterConverter, Filters

try:
    import enum

    _has_enum = True
except ImportError:
    _has_enum = False

log = logging.getLogger(__name__)


class BaseInterface:
    """
    Base class for all data model interfaces.
    Sub class it to implement your own interface for some data engine.
    """

    filter_converter_class = Type[BaseFilterConverter]
    """ when sub classing override with your own custom filter converter """

    """ Messages to display on CRUD Events """
    add_row_message = lazy_gettext("Added Row")
    edit_row_message = lazy_gettext("Changed Row")
    delete_row_message = lazy_gettext("Deleted Row")
    delete_integrity_error_message = lazy_gettext(
        "Associated data exists, please delete them first"
    )
    add_integrity_error_message = lazy_gettext(
        "Integrity error, probably unique constraint"
    )
    edit_integrity_error_message = lazy_gettext(
        "Integrity error, probably unique constraint"
    )
    general_error_message = lazy_gettext("General Error")

    database_error_message = lazy_gettext("Database Error")

    """ Tuple with message and text with severity type ex: ("Added Row", "info") """
    message = ()

    def __init__(self, obj: Type[Any]):
        self.obj = obj

    def __getattr__(self, name: str) -> Any:
        """
        Make mypy happy about the injected filters like self.datamodel.FilterEqual
        https://mypy.readthedocs.io/en/latest/cheat_sheet_py3.html#when-you-re-puzzled-or-when-things-are-complicated
        """
        return super().__getattr__(name)

    def _get_attr(self, col_name):
        if not hasattr(self.obj, col_name):
            # it's an inner obj attr
            try:
                _obj = self.obj
                for i in col_name.split("."):
                    try:
                        _obj = self.get_related_model(i)
                    except Exception:
                        _obj = getattr(_obj, i)
                return _obj
            except Exception:
                return None
        return getattr(self.obj, col_name)

    @staticmethod
    def _get_attr_value(item, col):
        if not hasattr(item, col):
            # it's an inner obj attr
            try:
                return reduce(getattr, col.split("."), item)
            except Exception:
                return ""
        if hasattr(getattr(item, col), "__call__"):
            # its a function
            return getattr(item, col)()
        else:
            # its an attribute
            value = getattr(item, col)
            # if value is an Enum instance than list and show widgets should display
            # its .value rather than its .name:
            if _has_enum and isinstance(value, enum.Enum):
                return value.value
            return value

    def get_filters(self, search_columns=None, search_filters=None):
        search_columns = search_columns or []
        return Filters(
            self.filter_converter_class,
            self,
            search_columns=search_columns,
            search_filters=search_filters,
        )

    def get_values_item(self, item, show_columns):
        return [self._get_attr_value(item, col) for col in show_columns]

    def _get_values(self, lst, list_columns):
        """
        Get Values: formats values for list template.
        returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]

        :param lst:
            The list of item objects from query
        :param list_columns:
            The list of columns to include
        """
        retlst = []
        for item in lst:
            retdict = {}
            for col in list_columns:
                retdict[col] = self._get_attr_value(item, col)
            retlst.append(retdict)
        return retlst

    def get_values(self, lst, list_columns):
        """
        Get Values: formats values for list template.
        returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]

        :param lst:
            The list of item objects from query
        :param list_columns:
            The list of columns to include
        """
        for item in lst:
            retdict = {}
            for col in list_columns:
                retdict[col] = self._get_attr_value(item, col)
            yield retdict

    def get_values_json(self, lst, list_columns):
        """
        Converts list of objects from query to JSON
        """
        result = []
        for item in self.get_values(lst, list_columns):
            for key, value in list(item.items()):
                if isinstance(value, datetime.datetime) or isinstance(
                    value, datetime.date
                ):
                    value = value.isoformat()
                    item[key] = value
                if isinstance(value, list):
                    item[key] = [str(v) for v in value]
            result.append(item)
        return result

    """
        Returns the models class name
        useful for auto title on views
    """

    @property
    def model_name(self):
        return self.obj.__class__.__name__

    """
        Next methods must be overridden
    """

    def query(
        self,
        filters=None,
        order_column="",
        order_direction="",
        page=None,
        page_size=None,
    ):
        pass

    def is_image(self, col_name):
        return False

    def is_file(self, col_name):
        return False

    def is_gridfs_file(self, col_name):
        return False

    def is_gridfs_image(self, col_name):
        return False

    def is_string(self, col_name):
        return False

    def is_text(self, col_name):
        return False

    def is_binary(self, col_name):
        return False

    def is_integer(self, col_name):
        return False

    def is_numeric(self, col_name):
        return False

    def is_float(self, col_name):
        return False

    def is_boolean(self, col_name):
        return False

    def is_date(self, col_name):
        return False

    def is_datetime(self, col_name):
        return False

    def is_enum(self, col_name):
        return False

    def is_relation(self, prop):
        return False

    def is_relation_col(self, col):
        return False

    def is_relation_many_to_one(self, prop):
        return False

    def is_relation_many_to_many(self, prop):
        return False

    def is_relation_one_to_one(self, prop):
        return False

    def is_relation_one_to_many(self, prop):
        return False

    def is_nullable(self, col_name):
        return True

    def is_unique(self, col_name):
        return False

    def is_pk(self, col_name):
        return False

    def is_pk_composite(self):
        raise False

    def is_fk(self, col_name):
        return False

    def get_max_length(self, col_name):
        return -1

    def get_min_length(self, col_name):
        return -1

    """
    -----------------------------------------
           FUNCTIONS FOR CRUD OPERATIONS
    -----------------------------------------
    """

    def add(self, item):
        """
        Adds object
        """
        raise NotImplementedError

    def edit(self, item):
        """
        Edit (change) object
        """
        raise NotImplementedError

    def delete(self, item):
        """
        Deletes object
        """
        raise NotImplementedError

    def get_col_default(self, col_name):
        pass

    def get_keys(self, lst):
        """
        return a list of pk values from object list
        """
        pk_name = self.get_pk_name()
        if self.is_pk_composite():
            return [[getattr(item, pk) for pk in pk_name] for item in lst]
        else:
            return [getattr(item, pk_name) for item in lst]

    def get_pk_name(self):
        """
        Returns the primary key name
        """
        raise NotImplementedError

    def get_pk_value(self, item):
        pk_name = self.get_pk_name()
        if self.is_pk_composite():
            return [getattr(item, pk) for pk in pk_name]
        else:
            return getattr(item, pk_name)

    def get(self, pk, filter=None):
        """
        return the record from key, you can optionally pass filters
        if pk exits on the db but filters exclude it it will return none.
        """
        pass

    def get_related_model(self, prop):
        raise NotImplementedError

    def get_related_interface(self, col_name):
        """
        Returns a BaseInterface for the related model
        of column name.

        :param col_name: Column name with relation
        :return: BaseInterface
        """
        raise NotImplementedError

    def get_related_obj(self, col_name, value):
        raise NotImplementedError

    def get_related_fk(self, model):
        raise NotImplementedError

    def get_columns_list(self):
        """
        Returns a list of all the columns names
        """
        return []

    def get_user_columns_list(self):
        """
        Returns a list of user viewable columns names
        """
        return self.get_columns_list()

    def get_search_columns_list(self):
        """
        Returns a list of searchable columns names
        """
        return []

    def get_order_columns_list(self, list_columns=None):
        """
        Returns a list of order columns names
        """
        return []

    def get_relation_fk(self, prop):
        pass