Enhancements: SQLA Interface, Mutation, InputObjectType and auto schema generation

[maquino1985]

Currently there are at least 3 critical features missing from the library:

Interface support, Mutation support, and InputObjectType support that integrate with SQLAlchemy models. A last, but nice to have, would be a type that automatically generates all graphene-sqlalchemy models and correctly assigns them the appropriate Interface type based on their model.

Interface is critically important for people using polymorphic database structures. The importance of auto-generated SQLAlchemy Mutations and InputObjectTypes is self explanatory.

a SQLAlchemyInterface should have as its meta fields a name and an SQLAlchemy model, e.g.

class BaseClassModelInterface(SQLAlchemyInterface):
    class Meta:
        name = 'BaseClassNode'
        model = BaseClassModel

Because it will act as a Node elsewhere, in my implementation I have it extend Node (but call super(AbstractNode) to specify it's meta rather than have it be overridden)

def exclude_autogenerated_sqla_columns(model: DeclarativeMeta) -> Tuple[str]:
    # always pull ids out to a separate argument
    autoexclude: List[str] = []
    for col in sqlalchemy.inspect(model).columns:
        if ((col.primary_key and col.autoincrement) or
                (isinstance(col.type, sqlalchemy.types.TIMESTAMP) and
                 col.server_default is not None)):
            autoexclude.append(col.name)
            assert isinstance(col.name, str)
    return tuple(autoexclude)

class SQLAlchemyInterfaceOptions(InterfaceOptions):
    model = None  #
    registry = None  #
    connection = None  #
    id = None  # type: str

class SQLAlchemyInterface(Node):
    @classmethod
    def __init_subclass_with_meta__(
            cls,
            model: DeclarativeMeta = None,
            registry: Registry = None,
            only_fields: Tuple[str] = (),
            exclude_fields: Tuple[str] = (),
            connection_field_factory: UnsortedSQLAlchemyConnectionField = default_connection_field_factory,
            **options
    ):
        _meta = SQLAlchemyInterfaceOptions(cls)
        _meta.name = f'{cls.__name__}Node'

        autoexclude_columns = exclude_autogenerated_sqla_columns(model=model)
        exclude_fields += autoexclude_columns

        assert is_mapped_class(model), (
            "You need to pass a valid SQLAlchemy Model in " '{}.Meta, received "{}".'
        ).format(cls.__name__, model)

        if not registry:
            registry = get_global_registry()

        assert isinstance(registry, Registry), (
            "The attribute registry in {} needs to be an instance of "
            'Registry, received "{}".'
        ).format(cls.__name__, registry)

        sqla_fields = yank_fields_from_attrs(
            construct_fields(
                obj_type=cls,
                model=model,
                registry=registry,
                only_fields=only_fields,
                exclude_fields=exclude_fields,
                connection_field_factory=connection_field_factory
            ),
            _as=Field
        )
        if not _meta:
            _meta = SQLAlchemyInterfaceOptions(cls)
        _meta.model = model
        _meta.registry = registry
        connection = Connection.create_type(
            "{}Connection".format(cls.__name__), node=cls)
        assert issubclass(connection, Connection), (
            "The connection must be a Connection. Received {}"
        ).format(connection.__name__)
        _meta.connection = connection
        if _meta.fields:
            _meta.fields.update(sqla_fields)
        else:
            _meta.fields = sqla_fields
        _meta.fields['id'] = graphene.GlobalID(cls, description="The ID of the object.")
        super(AbstractNode, cls).__init_subclass_with_meta__(_meta=_meta, **options)

    @classmethod
    def Field(cls, *args, **kwargs):  # noqa: N802
        return NodeField(cls, *args, **kwargs)

    @classmethod
    def node_resolver(cls, only_type, root, info, id):
        return cls.get_node_from_global_id(info, id, only_type=only_type)

    @classmethod
    def get_node_from_global_id(cls, info, global_id, only_type=None):
        try:
            node: DeclarativeMeta = info.context.get('session').query(cls._meta.model).filter_by(id=global_id).one_or_none()
            return node
        except Exception:
            return None

    @classmethod
    def from_global_id(cls, global_id):
        return global_id

    @classmethod
    def to_global_id(cls, type, id):
        return id

    @classmethod
    def resolve_type(cls, instance, info):
        if isinstance(instance, graphene.ObjectType):
            return type(instance)
        graphene_model = get_global_registry().get_type_for_model(type(instance))
        if graphene_model:
            return graphene_model
        else:
            raise ValueError(f'{instance} must be a SQLAlchemy model or graphene.ObjectType')

A mutation should take as its meta arguments the SQLAlchemy Model, it's CRUD operation . (Create Edit or Delete), and the graphene structure of its response (Output type)

class CreateFoos(SQLAlchemyMutation):
    class Arguments:
        foos = graphene.Argument(graphene.List(FooInput))

    class Meta:
        create = True
        model = FooModel
        structure = graphene.List

class SQLAlchemyMutation(graphene.Mutation):
    @classmethod
    def __init_subclass_with_meta__(cls, model=None, create=False,
                                    delete=False, registry=None,
                                    arguments=None, only_fields=(),
                                    structure: Type[Structure] = None,
                                    exclude_fields=(), **options):
        meta = SQLAlchemyMutationOptions(cls)
        meta.create = create
        meta.model = model
        meta.delete = delete

        if arguments is None and not hasattr(cls, "Arguments"):
            arguments = {}
            # don't include id argument on create
            if not meta.create:
                arguments['id'] = graphene.ID(required=True)

            # don't include input argument on delete
            if not meta.delete:
                inputMeta = type('Meta', (object,), {
                    'model': model,
                    'exclude_fields': exclude_fields,
                    'only_fields': only_fields
                })
                inputType = type(cls.__name__ + 'Input',
                                 (SQLAlchemyInputObjectType,),
                                 {'Meta': inputMeta})
                arguments = {'input': inputType(required=True)}
        if not registry:
            registry = get_global_registry()
        output_type: ObjectType = registry.get_type_for_model(model)
        if structure:
            output_type = structure(output_type)
        super(SQLAlchemyMutation, cls).__init_subclass_with_meta__(_meta=meta, output=output_type, arguments=arguments, **options)

    @classmethod
    def mutate(cls, info, **kwargs):
        session = get_session(info.context)
        with session.no_autoflush:
            meta = cls._meta
            model = None

            if meta.create:
                model = meta.model(**kwargs['input'])
                session.add(model)
            else:
                model = session.query(meta.model).filter(meta.model.id == kwargs['id']).first()

            if meta.delete:
                session.delete(model)
            else:

                def setModelAttributes(model, attrs):
                    relationships = model.__mapper__.relationships
                    for key, value in attrs.items():
                        if key in relationships:
                            if getattr(model, key) is None:
                                # instantiate class of the same type as
                                # the relationship target
                                setattr(model, key,
                                        relationships[key].mapper.entity())
                            setModelAttributes(getattr(model, key), value)
                        else:
                            setattr(model, key, value)

                setModelAttributes(model, kwargs['input'])
            session.commit()
            return model

    @classmethod
    def Field(cls, *args, **kwargs):
        return graphene.Field(cls._meta.output,
                              args=cls._meta.arguments,
                              resolver=cls._meta.resolver)

an SQLAlchemy InputObjectType should introspect the sqla model and autogenerate fields to select based upon and set the appropriate field data type:

e.g.

class Bar(SQLAlchemyInputObjectType):
    class Meta:
        model = BarModel
        exclude_fields = ( 'polymorphic_discriminator', 'active', 'visible_id')

```python
class SQLAlchemyInputObjectType(InputObjectType):
    @classmethod
    def __init_subclass_with_meta__(
            cls,
            model=None,
            registry=None,
            skip_registry=False,
            only_fields=(),
            exclude_fields=(),
            connection=None,
            connection_class=None,
            use_connection=None,
            interfaces=(),
            id=None,
            connection_field_factory=default_connection_field_factory,
            _meta=None,
            **options
    ):
        autoexclude = []

        # always pull ids out to a separate argument
        for col in sqlalchemy.inspect(model).columns:
            if ((col.primary_key and col.autoincrement) or
                    (isinstance(col.type, sqlalchemy.types.TIMESTAMP) and
                     col.server_default is not None)):
                autoexclude.append(col.name)

        if not registry:
            registry = get_global_registry()
        sqla_fields = yank_fields_from_attrs(
            construct_fields(cls, model, registry, only_fields, exclude_fields + tuple(autoexclude), connection_field_factory),
            _as=Field,
        )
        # create accessor for model to be retrieved for querying
        cls.sqla_model = model
        if use_connection is None and interfaces:
            use_connection = any(
                (issubclass(interface, Node) for interface in interfaces)
            )

        if use_connection and not connection:
            # We create the connection automatically
            if not connection_class:
                connection_class = Connection

            connection = connection_class.create_type(
                "{}Connection".format(cls.__name__), node=cls
            )

        if connection is not None:
            assert issubclass(connection, Connection), (
                "The connection must be a Connection. Received {}"
            ).format(connection.__name__)

        for key, value in sqla_fields.items():
            if not (isinstance(value, Dynamic) or hasattr(cls, key)):
                setattr(cls, key, value)

        super(SQLAlchemyInputObjectType, cls).__init_subclass_with_meta__(**options)

[maquino1985]

Lastly, a nice to have feature would auto generate the schema and query for you, given a list of models, interfaces, excluded models, and a default node interface (because we'll usually be overriding Node with a custom node that provides the database ID anyways)

e.g.

class Query(SQLAlchemyAutoSchemaFactory):
    class Meta:
        interfaces = (FooInterface, BarInterface,)
        models = (*foos, *bars, *others, *associatives) # associatives are for AssociationObjectTypes, need to create the graphene models for these to work
        node_interface = CustomNode
        excluded_models = (TrackedEntityModel, BaseControlledVocabulary)

class SQLAlchemyAutoSchemaFactory(graphene.ObjectType):

    @staticmethod
    def set_fields_and_attrs(klazz, node_model, field_dict):
        _name = to_snake_case(node_model.__name__)
        field_dict[f'all_{(pluralize_name(_name))}'] = FilteredConnectionField(node_model)
        field_dict[_name] = node_model.Field()
        # log.info(f'interface:{node_model.__name__}')
        setattr(klazz, _name, node_model.Field())
        setattr(klazz, "all_{}".format(pluralize_name(_name)), FilteredConnectionField(node_model))

    @classmethod
    def __init_subclass_with_meta__(
            cls,
            interfaces: Tuple[Type[SQLAlchemyInterface]] = (),
            models: Tuple[Type[DeclarativeMeta]] = (),
            excluded_models: Tuple[Type[DeclarativeMeta]] = (),
            node_interface: Type[Node] = Node,
            default_resolver: ResolveInfo = None,
            _meta=None,
            **options
    ):
        if not _meta:
            _meta = ObjectTypeOptions(cls)

        fields = OrderedDict()

        for interface in interfaces:
            if issubclass(interface, SQLAlchemyInterface):
                SQLAlchemyAutoSchemaFactory.set_fields_and_attrs(cls, interface, fields)
        for model in excluded_models:
            if model in models:
                models = models[:models.index(model)] + models[models.index(model) + 1:]
        possible_types = ()
        for model in models:
            model_name = model.__name__
            _model_name = to_snake_case(model.__name__)

            if hasattr(cls, _model_name):
                continue
            if hasattr(cls, "all_{}".format(pluralize_name(_model_name))):
                continue
            for iface in interfaces:
                if issubclass(model, iface._meta.model):
                    model_interface = (iface,)
                    break
            else:
                model_interface = (CustomNode,)

            _node_class = type(model_name,
                               (SQLAlchemyObjectType,),
                               {"Meta": {"model": model, "interfaces": model_interface, "only_fields": []}})
            fields["all_{}".format(pluralize_name(_model_name))] = FilteredConnectionField(_node_class)
            setattr(cls, "all_{}".format(pluralize_name(_model_name)), FilteredConnectionField(_node_class))
            fields[_model_name] = node_interface.Field(_node_class)
            setattr(cls, _model_name, node_interface.Field(_node_class))
            possible_types += (_node_class,)
        if _meta.fields:
            _meta.fields.update(fields)
        else:
            _meta.fields = fields
        _meta.schema_types = possible_types

        super(SQLAlchemyAutoSchemaFactory, cls).__init_subclass_with_meta__(_meta=_meta, default_resolver=default_resolver, **options)

    @classmethod
    def resolve_with_filters(cls, info: ResolveInfo, model: Type[SQLAlchemyObjectType], **kwargs):
        query = model.get_query(info)
        for filter_name, filter_value in kwargs.items():
            model_filter_column = getattr(model._meta.model, filter_name, None)
            if not model_filter_column:
                continue
            if isinstance(filter_value, SQLAlchemyInputObjectType):
                filter_model = filter_value.sqla_model
                q = FilteredConnectionField.get_query(filter_model, info, sort=None, **kwargs)
                # noinspection PyArgumentList
                query = query.filter(model_filter_column == q.filter_by(**filter_value))
            else:
                query = query.filter(model_filter_column == filter_value)
        return query

[jnak]

Hello @maquino1985 ,

Thanks for starting these 4 enhancement proposals. There is quite a bit to unpack and discuss about each of those enhancements. Since those enhancements seem to be fairly independent from one another, can you create separate issues in order for the discussions to be productive and clear?

For example, I agree with you that there is potential for graphene-sqlalchemy to help with mutations but it is not obvious to me why your suggested implementation is the right solution. You can see some of my questions here. In addition to those questions, I would also like to hear your thoughts on input validation since it's very uncommon for a mutation to save data without validating user inputs.

Given that we have only a couple of semi-active maintainers for this project, I would also suggest that you start with the one issue you care the most about so we can make good progress on it instead making slower progress on all of those. There is only so much time per week the maintainers can dedicate to this project.

I hope you understand.
Best.

[maquino1985]

Hi @jnak , can you elaborate on your question about input validation? Other than requiring specific inputs of specific types in the Mutation Arguments, input validation should always occur in the SQLAlchemy model and/or at the database level upon updating or saving any data. I

[github-actions]

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related topics referencing this issue.