Thoughts on lower-level internals

[dwreeves]

I've been playing around with the idea that it would be much easier to rework the internals to achieve a few objectives:

Define terms clearly

SQLAlchemy seems to be inconsistent, or at least confusing, on what it refers to as "properties" versus "attributes". Simple example below:

from sqlmodel import SQLModel, Field
from sqlalchemy import inspect as sqla_inspect

class Boathouse(SQLModel, table=True):
    id: int = Field(primary_key=True)
    name: str = Field(sa_column_kwargs=dict(unique=True))

print("attrs['id'] type:        ", type(sqla_inspect(Boathouse).attrs["id"]))
print("attrs['id'].is_attribute:", sqla_inspect(Boathouse).attrs["id"].is_attribute)
print("attrs['id'].is_property: ", sqla_inspect(Boathouse).attrs["id"].is_property)
print("---")
print("model.id type:           ", type(Boathouse.id))
print("model.id.is_attribute:   ", Boathouse.id.is_attribute)
print("model.id.is_property:    ", Boathouse.id.is_property)

The output of the above code is:

attrs['id'] type:         <class 'sqlalchemy.orm.properties.ColumnProperty'>
attrs['id'].is_attribute: False
attrs['id'].is_property:  True
---
model.id type:            <class 'sqlalchemy.orm.attributes.InstrumentedAttribute'>
model.id.is_attribute:    True
model.id.is_property:     False

Even though the mapping sqla_inspect(Boathouse).attrs seems to imply the mapped objects are "attributes," it ends up returning "properties" instead, and the is_attribute / is_property attributes make that clear.

I propose that Union[str, InstrumentedAttribute] should always refer to an "attr", and that Union[ColumnProperty, RelationshipProperty] should always refer to a "prop".

Right now this is not entirely the case. The most clear example is that ModelAdmin.get_model_attr() returns a property, so it should be renamed to get_model_prop().

Additionally, this library uses a lot of powerful OOP libraries that have potentially overlapping concepts. We can improve readability (most notably in contexts where things are being mixed and matched) by prefixing SQLAlchemy variables with sqla_*, and prefixing WTForms variables with wtf_*. So instead of "model", it's a sqla_model; instead of a field, it's a wtf_field.

Clear converter function signatures

I pulled the design pattern from Flask-Admin, and added my own typing:

from typing import TypeVar
from typing_extensions import Protocol

T_MP = TypeVar("T_MP", ColumnProperty, RelationshipProperty)

class ConverterCallback(Protocol):
    def __call__(
            self,
            model: type,
            prop: T_MP,
            kwargs: Dict[str, Any]
    ) -> UnboundField:
        ...

T_CC = TypeVar("T_CC", bound=ConverterCallback)

def converts(*args: str) -> Callable[[T_CC], T_CC]:
    def _inner(func: T_CC) -> T_CC:
        func._converter_for = frozenset(args)
        return func
    return _inner

Again on naming consistency: the "converter" should be the object typed as ModelConverterBase. The methods are not converters, they are converter callbacks.

This also means, by the way, that the method ModelConverterBase.get_converter() should be renamed to ModelConverterBase.get_converter_callback().

"The function signature is more limited-- How do I do X?" Here is an example of why, for example, the column: Column input is not needed. Here is a refactor of the converter callback for Enum types:

class ModelConverter(ModelConverterBase):

    # ...

    @converts("Enum")
    def conv_Enum(
            self, model: type, prop: ColumnProperty, kwargs: Dict[str, Any]
    ) -> UnboundField:
        available_choices = [(e, e) for e in prop.columns[0].type.enums]
        accepted_values = [choice[0] for choice in available_choices]

        kwargs["choices"] = available_choices
        kwargs.setdefault("validators", [])
        kwargs["validators"].append(validators.AnyOf(accepted_values))
        kwargs["coerce"] = lambda v: v.name if isinstance(v, Enum) else str(v)
        return SelectField(**kwargs)

In this case, access to the column is done through prop.columns[0].

ModelConverterBase.convert() does too much.

This method not only converts the type, it also adds kwargs that can be processed upstream.

The convert method should be simple, and kwargs should be handled upstream of convert. The only code below that wasn't implemented is get_callback(). Everything else is unabridged.

class ModelConverterBase:

    _callbacks: Dict[str, ConverterCallable] = {}

    def __init__(self):
        super().__init__()
        self._register_callbacks()

    def _register_callbacks(self):
        callbacks: Dict[str, ConverterCallable] = {}
        for name in dir(self):
            obj = getattr(self, name)
            if hasattr(obj, "_converter_for"):
                for classname in obj._converter_for:
                    callbacks[classname] = obj
        self._callbacks = callbacks

    def get_callback(
            self,
            model: type,
            prop: T_MP
    ) -> ConverterCallable:
        ...

    def convert(
            self,
            model: type,
            prop: T_MP,
            kwargs: Mapping[str, Any]
    ) -> UnboundField:
        converter = self.get_callback(model, prop)
        return converter(
            model=model,
            prop=prop,
            kwargs=kwargs
        )

Handle all parsing by iterating over Lists of AdminAttributes.

I think instead of working directly with SQLA objects and WTF objects, these should be given an extra layer of abstraction that relates them to the admin panel object. These should be called AdminAttributes. This class addresses a lot of the aforementioned problems by creating a common interface to do various tasks:

• interacting with "props" and "attrs" from SQLAlchemy
• relating SQLA and WTF objects and concepts.
• storing data such as the "label" and "form_args" next to their corresponding fields (they are defined in separate dicts by the Flask-Admin API, but they should be all placed in a single object).
• all or nearly all processing of wtforms.Field kwargs can be handled in AdminAttribute.
• many namespaced concepts (such as "prop" and "attr") can be source-of-truth'd in the AdminAttribute

In this context, the main parsing challenge just becomes turning a SQLAlchemy model's fields into AdminAttributes as defined by the ModelAdmin's classvars. (But that is easy.)

Here is the code I have so far for AdminAttributes:

class AdminAttribute:

    def __init__(
            self,
            sqla_model: type,
            sqla_attribute: Union[str, InstrumentedAttribute],
            *,
            label: Optional[str] = None,
            wtf_field_type_override: Type[Field] = None,
            wtf_extra_field_kwargs: Dict[str, Any] = None,
            wtf_extra_validators: List[Callable[[Form, Field], None]] = None
    ):
        self.sqla_model = sqla_model

        if isinstance(sqla_attribute, str):
            self.sqla_attribute = getattr(sqla_model, sqla_attribute)
        else:
            self.sqla_attribute = sqla_attribute
        assert isinstance(self.sqla_attribute, InstrumentedAttribute)
        assert self.sqla_attribute.class_ is sqla_model

        self.label = label

        self.wtf_extra_field_kwargs = wtf_extra_field_kwargs or {}
        self.wtf_field_type_override = wtf_field_type_override
        self.wtf_extra_validators = wtf_extra_validators or []

    @property
    def is_relationship(self) -> bool:
        return isinstance(self.sqla_attribute.prop, RelationshipProperty)

    @property
    def ensured_label(self) -> str:
        """self.label is allowed to be optional.
        If you definitely want a string to represent the column,
        then you want to use this property.
        """
        return self.label or self.sqla_property.key

    @property
    def sqla_property(self) -> MapperProperty:
        return self.sqla_attribute.prop

    @cached_property
    def sqla_column(self) -> Column:
        assert len(self.sqla_property.columns) == 1, "Multiple-column properties not supported"
        column = self.sqla_property.columns[0]
        return column

    def _get_default_value(self) -> Any:
        default = getattr(self.sqla_column, "default", None)

        if default is not None:
            # Only actually change default if it has an attribute named
            # 'arg' that's callable.
            callable_default = getattr(default, "arg", None)

            if callable_default is not None:
                # ColumnDefault(val).arg can be also a plain value
                default = (
                    callable_default(None)
                    if callable(callable_default)
                    else callable_default
                )

        return default

    @property
    def wtf_base_validators(self) -> List[Validator]:
        if self.is_relationship:
            return []
        li = []
        if self.sqla_column.nullable:
            li.append(validators.Optional())
        else:
            li.append(validators.InputRequired())
        return li

    @property
    def _wtf_default_field_kwargs(self) -> Dict[str, Any]:
        kwargs = {
            "label": self.label,
            "validators": [
                *self.wtf_base_validators,
                *self.wtf_extra_validators
            ],
            "description": self.sqla_property.doc,
            "render_kw": {"class": "form-control"},
            "default": self._get_default_value()
        }
        if self.is_relationship:
            allow_blank = True
            for pair in self.sqla_property.local_remote_pairs:
                if not pair[0].nullable:
                    allow_blank = False
            kwargs["allow_blank"] = allow_blank
        return kwargs

    async def get_object_list(
            self,
            engine: Union[Engine, AsyncEngine]
    ) -> List[Any]:
        target_model = self.sqla_property.mapper.class_
        pk = sqlalchemy_inspect(target_model).primary_key[0].name
        stmt = select(target_model)

        if isinstance(engine, Engine):
            with Session(engine) as session:
                objects = await anyio.to_thread.run_sync(session.execute, stmt)
                object_list = [
                    (str(getattr(obj, pk)), obj)
                    for obj in objects.scalars().all()
                ]
        else:
            async with AsyncSession(engine) as session:
                objects = await session.execute(stmt)
                object_list = [
                    (str(getattr(obj, pk)), obj)
                    for obj in objects.scalars().all()
                ]
        return object_list

    @cached_property
    def wtf_field_kwargs(self) -> ChainMap:
        d = ChainMap(
            self.wtf_extra_field_kwargs,
            self._wtf_default_field_kwargs,
        )
        return d

And here is the code I have for parsing ModelAdmin classvars into List[AdminAttribute]s:

class ModelAdmin(BaseModelAdmin, metaclass=ModelAdminMeta):

    # ... more here ...

    def __init__(self) -> None:

        self._admin_attributes: Dict[Union[str, InstrumentedAttribute], AdminAttribute] = {}

        self._list_admin_attributes = [
            self.get_admin_attr(key) for key in self.column_list
        ]

        # ... more here ...

    def get_admin_attr(self, key: Union[str, InstrumentedAttribute]) -> AdminAttribute:
        val = self._admin_attributes.get(key)
        if val is None:
            self._set_admin_attr(key)
            val = self._admin_attributes.get(key)
        return val

    @staticmethod
    def _get_key_and_ia(
            model: type,
            v: Union[str, InstrumentedAttribute]
    ) -> Tuple[str, InstrumentedAttribute]:
        if isinstance(v, str):
            return v, getattr(model, v)
        elif isinstance(v, InstrumentedAttribute):
            return v.key, v
        else:
            raise TypeError(
                f"Input must be str or InstrumentedAttribute; received {type(v)!r}."
            )

    def _lookup_sqla_column_key(
            self,
            mapping: Dict[Union[str, InstrumentedAttribute], T],
            key: Union[str, InstrumentedAttribute]
    ) -> T:
        """Does a lookup on a dict with `Union[str, InstrumentedAttribute]` keys.
        The point is that this works even if the key lookup and key storage types are
        mismatched.
        """
        if key in mapping:
            return mapping[key]
        if isinstance(key, str):
            key2: InstrumentedAttribute = getattr(self.model, key)
            return mapping.get(key2)
        elif isinstance(key, InstrumentedAttribute):
            key2 = key.key
            return mapping.get(key2)
        else:
            raise TypeError(
                f"Input must be str or InstrumentedAttribute; received {type(v)!r}."
            )

    def _set_admin_attr(self, key: Union[str, InstrumentedAttribute]) -> None:
        """Doesn't ever need to be directly called."""
        key, ia = self._get_key_and_ia(self.model, key)

        attr = AdminAttribute(
            sqla_model=self.model,
            sqla_attribute=key,
            label=self._lookup_sqla_column_key(self.column_labels, key),
            wtf_field_type_override=self._lookup_sqla_column_key(self.form_overrides, key),
            wtf_extra_field_kwargs=self._lookup_sqla_column_key(self.form_args, key)
        )

        self._admin_attributes[key] = attr
        self._admin_attributes[ia] = attr

[dwreeves]

Let me know what you think.

The tldr is-- I want to have all the parsing be handled by something like a AdminAttribute. I think it will dramatically reduce the complexity of the code and make it much easier to implement additional features.

[aminalaee]

Hey @dwreeves , thanks for taking the time and thinking about these. You touched on a few points, here are my quick thoughts:

Define terms clearly

Yes, that's correct. I've sometimes mixed and matched columns, attributes and properties, and I've tried to improve on that, but it's still not very clear. Some part of that is from the SQLAlchemy I think which doesn't make this clear. Definitely worth improving.

But about prefixing form fields and attributes, honestly I think that's an overkill right now, because in the context of this libraries field is always part of the form and SQLAlchemy won't use that, and column/attribute/property is part of the SQLAlchemy model. What do you think?

Clear converter function signatures
And

ModelConverterBase.convert() does too much.

Actually this part in both SQLAdmin and Flask-Admin is heavily forked from WTForms-SQLAlchemy and I haven't really got a chance to improve on that. It's not really readable or even modern python because they have to support a wider range of Python. Definitely this can be improved.

Handle all parsing by iterating over Lists of AdminAttributes.

This one is the most serious one and I'm really hesitant because it's a much larger effort. I'd suggest we can start by baby steps and revise this along the way. We can maybe create separate issues for whatever we agree to improve and work on them separately. Does that make sense?

Again, thanks a lot for taking the time and contributing to the project. It's very much appreciated.

[dwreeves]

I'll defer to what you want on prefixes for things like sqla_model or wtf_form_type. I feel like prop or property might still need to be prefixed, though.

I have a pull request that starts to tackle some of those things. #115 I wonder if I took too big of a step, though? Maybe not baby-stepped enough.

[dwreeves]

Let me know what you think on the PR, and please push back. It's a proposal for what would end up being a very significant change to the internals, so shouldn't be taken lightly and you shouldn't approve it if you are unsure of the design decisions.

[dwreeves]

Hey @aminalaee, do you have any intentions on looking into #116? I want to continue working on sqladmin while I have time (I won't be unemployed forever!), and #116 is blocking other steps.

[aminalaee]

Hey @dwreeves ,
Yes I've been meaning to review that for some time, it's a bit large and I haven't had a chance, but sure I'll do that.