Docs or instructions to deal with Many to many fields

[ccsv]

Are there any examples of how to deal with Manytomany fields?

Do I have to write a query to prefetch choice fields when updating the fields?

[maxmorlocke]

Similar to the deletion, I created manytomany resolver helpers. Again, this is stuff lacking testing and documentation on how to use (and possibly proprietary code) so it hasn't been added to the project. The good part is this makes adding a new set of m2m relationships easy. The bad part is there's a bunch of undocumented dependencies like adding a mixin to a model and making sure specific properties are set on said model to make the magic work. This works just fine in practice, but is problematic as heck when onboarding a new dev.

from django.core.exceptions import ImproperlyConfigured
from django.db import transaction
from resetbutton.graphql.convert_kwargs_to_snake_case import convert_to_snake_case


"""
GraphQL Resolver to manage adding existing objects to many-to-many relationships.
Assumes the parent object is where the object where the many to many relationship
is defined (i.e. this doesn't work on reverse relationships).

This resolver assumes that underlying model defines AddManyToManyModelByIdBaseMixin
subclasses for each related field, which in turn provides a `add_{fieldname}_by_id` method.
"""


class AdditionManyToManyResolver:
    model_lookup_field = "id"
    m2m_relation_field = None

    def __init__(self, info=None, id=None, related_id=None, related_typename=None, m2m_through_values=None):
        required_arguments = ["model_lookup_field", "m2m_relation_field"]
        required_fields_missing = []
        for required_argument in required_arguments:
            if getattr(self, required_argument, None) is None:
                required_fields_missing.append(required_argument)

        if required_fields_missing:
            raise ImproperlyConfigured(f"Need to define {required_fields_missing}")

        self.graphql_info = info
        self.model_lookup_value = id
        self.related_object_id = related_id
        self.related_typename = related_typename
        self.m2m_through_values = m2m_through_values

    def get_queryset(self):
        raise NotImplementedError("Developer needs to build this...")

    def get_parent_instance(self):
        raise NotImplementedError("Developer needs to build this...")

    def get_object(self):
        qs = self.get_queryset()
        return qs.get(id=self.related_object_id)

    @transaction.atomic
    def add(self, *args, **kwargs):
        parent = self.get_parent_instance()
        child = self.get_object()
        method_name = f"add_{self.m2m_relation_field}_by_id"
        try:
            add_to_parent = getattr(parent, method_name)
        except AttributeError as e:
            raise ImproperlyConfigured(
                f"Could not find {method_name} method. The parent model must implement a "
                "AddManyToManyModelByIdBaseMixin subclass for the "
                f"{self.m2m_relation_field} relationship."
            ) from e

        add_to_parent([child.id], m2m_through_values=[convert_to_snake_case(self.m2m_through_values)])
        return parent

from django.core.exceptions import ImproperlyConfigured, ObjectDoesNotExist
from django.db.models import ProtectedError


class DeletionManyToManyResolver:
    model_lookup_field = "id"
    m2m_relation_field = None

    def __init__(self, info=None, id=None, related_id=None, related_typename=None):
        required_arguments = ["model_lookup_field", "m2m_relation_field"]
        required_fields_missing = []
        for required_argument in required_arguments:
            if getattr(self, required_argument, None) is None:
                required_fields_missing.append(required_argument)

        if required_fields_missing:
            raise ImproperlyConfigured(f"Need to define {required_fields_missing}")

        self.graphql_info = info
        self.model_lookup_value = id
        self.related_object_id = related_id
        self.related_typename = related_typename

    def get_queryset(self):
        parent_object = self.get_parent_instance()
        m2m_field = parent_object.get_m2m_field(self.m2m_relation_field)
        return m2m_field.all()

    def get_parent_instance(self):
        raise NotImplementedError("Developer needs to build this...")

    def get_object(self):
        qs = self.get_queryset()
        try:
            lookup_dict = self.get_lookup_dict()
            instance = qs.get(**lookup_dict)
        except ObjectDoesNotExist:
            instance = None
        return instance

    def destroy(self, *args, **kwargs):
        instance = self.get_object()
        original_instance = instance
        parent_object = self.perform_destroy(instance)
        setattr(original_instance, self.model_lookup_field, self.model_lookup_value)
        return parent_object

    def perform_destroy(self, instance):
        if instance:
            parent_object = self.get_parent_instance()
            m2m_field = parent_object.get_m2m_field(self.m2m_relation_field)
            if not m2m_field.exists():
                return

            m2m_field.remove(instance)
            parent_object.save()

            try:
                instance.delete()
            except ProtectedError:
                pass  # other references floating around, so ignore

            return parent_object

    def get_lookup_dict(self):
        return {self.model_lookup_field: self.model_lookup_value}

from django.db import transaction

from core.graphql.resolvers.mixins.serializer_mutation_helper import SerializerMutationHelperMixin
from resetbutton.graphql.convert_kwargs_to_snake_case import convert_to_snake_case


class SerializerManyToManyMutationResolver(SerializerMutationHelperMixin):
    serializer_class = None
    partial = None
    model_lookup_field = None
    include_person_in_context = False
    m2m_relation_field = None
    m2m_through_values_required = False
    info = None
    input = None
    related_id = None

    def __init__(self, info=None, input=None, related_id=None, related_typename=None, m2m_through_values=None):
        if self.m2m_through_values_required and m2m_through_values is None:
            raise TypeError("Missing required keyword argument: 'm2m_through_values'")

        self.verify_required_fields(
            [
                "serializer_class",
                "partial",
                "model_lookup_field",
                "m2m_relation_field",
            ]
        )
        self.graphql_info = info
        self.input_data = self.clean_input_data(input)
        self.related_object_lookup_value = related_id
        self.related_typename = related_typename
        self.m2m_through_values = m2m_through_values

    def get_queryset(self):
        parent_object = self.get_parent_instance()
        m2m_field = parent_object.get_m2m_field(self.m2m_relation_field)
        return m2m_field.all()

    def get_parent_instance(self):
        raise NotImplementedError("Developer needs to build this...")

    def get_instance(self):
        if self.input_data.get(self.model_lookup_field):
            lookup_dict = {self.model_lookup_field: self.input_data[self.model_lookup_field]}
            queryset = self.get_queryset()
            instance = queryset.get(**lookup_dict)
        else:
            instance = None

        return instance

    def perform_related_mutation(self):
        instance = self.get_instance()
        context = self.get_context()

        serializer = self.serializer_class(instance, data=self.input_data, context=context, partial=self.partial)
        serializer.is_valid(raise_exception=True)
        mutated_instance = serializer.save()
        return mutated_instance

    @transaction.atomic
    def perform_mutation(self, *args, **kwargs):
        related_object = self.perform_related_mutation()

        parent_object = self.get_parent_instance()
        m2m_field = parent_object.get_m2m_field(self.m2m_relation_field)

        m2m_field.remove(related_object)  # so we can re-populate the through values when updating
        m2m_field.add(related_object, through_defaults=convert_to_snake_case(self.m2m_through_values))

        rd = parent_object.get_managed_field_dict(self.m2m_relation_field)
        if rd["boolean_field_name"]:
            setattr(parent_object, rd["boolean_field_name"], False)

        parent_object.save()

        return parent_object


from django.core.exceptions import ImproperlyConfigured

from resetbutton.graphql.convert_kwargs_to_snake_case import convert_to_snake_case


class SerializerMutationHelperMixin:
    def verify_required_fields(self, required_fields):
        required_fields_missing = []
        for required_field in required_fields:
            if getattr(self, required_field, None) is None:
                required_fields_missing.append(required_field)

        if required_fields_missing:
            raise ImproperlyConfigured(f"Need to define {required_fields_missing}")

    def get_context(self):
        context = {"request": self.graphql_info.context["request"]}
        if self.include_person_in_context:
            context["person"] = self.graphql_info.context["request"].user.filingperson
        return context

    def clean_input_data(self, input_data):
        return convert_to_snake_case(input_data)

Note that this stuff requires a mixin on the models as well:

from django.core.exceptions import ImproperlyConfigured, ObjectDoesNotExist
from django.db import transaction


class AddManyToManyModelByIdBaseMixin:
    def get_managed_field_dict(self, field_name):
        """
        Retrieve dictionary from CustomMeta.m2m_managed_relations where m2m_field_name is provided field_name arg
        """
        try:
            result = list(filter(lambda rd: rd["m2m_field_name"] == field_name, self.CustomMeta.m2m_managed_relations))[
                0
            ]
        except IndexError:
            raise ImproperlyConfigured(f"Missing managed relation for {field_name}")
        return result

    def get_m2m_field(self, field_name):
        """
        Retrieves the m2m field (e.g. cosigners) from the model that corresponds with this name based on
        the provided data in CustomMeta.m2m_managed_relations
        """
        rd = self.get_managed_field_dict(field_name)
        m2m_field_name = rd["m2m_field_name"]
        return getattr(self, m2m_field_name)

    def get_m2m_values_queryset(self, field_name):
        """
        Gets all possible values for this m2m field based on CustomMeta.m2m_managed_relations 'values_queryset'
        """
        rd = self.get_managed_field_dict(field_name)
        qs_fxn = getattr(self, rd["values_queryset"])
        return qs_fxn()

    def extract_key(self, unknown_format_id):
        try:
            pk = unknown_format_id["id"]
        except TypeError:
            pk = unknown_format_id

        return pk

    def get_m2m_object(self, primary_key, field_name):
        return self.get_m2m_values_queryset(field_name).get(pk=primary_key)

    def get_objects(self, object_ids, field_name, field_name_singular):
        """
        Converts list of object_ids to actual objects
        """
        objects = []
        errors = {}
        for object_id in object_ids:
            try:
                primary_key = self.extract_key(object_id)
            except KeyError:
                errors["id"] = f"All {field_name_singular} objects must have an ID."
                continue

            try:
                obj = self.get_m2m_object(primary_key, field_name)
                objects.append(obj)
            except ObjectDoesNotExist:
                errors[primary_key] = f"{field_name_singular} with id {primary_key} does not exist."

        if errors:
            raise ValueError(errors)

        return objects

    @transaction.atomic
    def replace_all(self, new_objects, field_name, m2m_through_values=None):
        """
        Replaces all entries in m2m field with new objects
        """
        m2m_field = self.get_m2m_field(field_name)
        old_objects = list(m2m_field.all())

        # we need to use `.add` rather than `.set` to allow different
        # through values for each individual new object
        m2m_field.set([])
        for i, new_object in enumerate(new_objects):
            try:
                through_defaults = m2m_through_values[i]
            except (IndexError, TypeError):
                through_defaults = None

            m2m_field.add(new_object, through_defaults=through_defaults)

        for old_object in old_objects:
            self.delete_protected(old_object)
        self.save()

    @transaction.atomic()
    def update_m2m_field(self, new_objects, field_name, replace_all, m2m_through_values=None):
        if replace_all:
            self.replace_all(new_objects, field_name, m2m_through_values=m2m_through_values)
        else:
            m2m_field = self.get_m2m_field(field_name)
            for i, obj in enumerate(new_objects):
                try:
                    through_defaults = m2m_through_values[i]
                except (IndexError, TypeError):
                    through_defaults = None
                m2m_field.remove(obj)  # to repopulate through_defaults
                m2m_field.add(obj, through_defaults=through_defaults)

        self.save()

    def get_m2m_expected_boolean_field(self, field_name):
        """
        Gets the m2m boolean field (e.g. has_no_cosigners).  Note - should return the boolean value
        """
        rd = self.get_managed_field_dict(field_name)
        return getattr(self, rd["boolean_field_name"])

    def set_m2m_expected_boolean_field(self, field_name):
        """
        We want to protect ourselves very carefully here.  This boolean should
        be default True, indicating is has no [m2m field].  If a user has set it to False
        and deletes all related objects, they should explicitly update the boolean.
        """
        expected_boolean = self.get_m2m_expected_boolean_field(field_name)
        rd = self.get_managed_field_dict(field_name)
        # This allows legacy values like "cofiling" on BankruptcyCases to work w/o changing f/e code.
        expected_m2m_boolean_value = False if rd.get("legacy_boolean_behavior", False) is True else True

        if expected_boolean is expected_m2m_boolean_value:
            has_values = self.get_m2m_values_queryset(field_name).exists()
            m2m_value = has_values if rd.get("legacy_boolean_behavior", False) is True else not has_values
            setattr(self, rd["boolean_field_name"], m2m_value)
        self.save()

With some reference implementations:

from ariadne import gql

from core.graphql.resolvers.addition_many_to_many_resolver import AdditionManyToManyResolver
from core.graphql.resolvers.deletion_many_to_many_resolver import DeletionManyToManyResolver
from core.graphql.resolvers.mark_no_many_to_many_resolver import MarkNoManyToManyResolver
from core.graphql.resolvers.serializer_many_to_many_mutation_resolver import SerializerManyToManyMutationResolver
from core.serializers.other_person import OtherPersonSerializer
from filing.graphql.types.resolve import get_queryset_by_typename
from resetbutton.graphql.decorators.assess_state_required import assess_state_required
from resetbutton.graphql.decorators.login_required import login_required
from resetbutton.graphql.mutation_type import mutation


type_defs = gql(
    """
    extend type Mutation {
        cosigner(
            relatedId: UUID!,
            relatedTypename: String!,
            otherPersonInput: OtherPersonInput!
        ): Cosignable!

        addCosigner(
            relatedId: UUID!,
            relatedTypename: String!,
            otherPersonId: UUID!
        ): Cosignable!

        deleteCosigner(cosignerId: UUID!, relatedId: UUID!, relatedTypename: String!): Cosignable!

        markNoCosigners(id: UUID!, typename: String!): Cosignable!
    }
    """
)


class CosignerMutationResolver(SerializerManyToManyMutationResolver):
    serializer_class = OtherPersonSerializer
    partial = False
    include_person_in_context = True
    m2m_relation_field = "cosigners"
    model_lookup_field = "id"

    def get_parent_instance(self):
        qs = get_queryset_by_typename(self.graphql_info.context["request"].user, self.related_typename)
        return qs.get(pk=self.related_object_lookup_value)

    @mutation.field("cosigner")
    @login_required()
    @assess_state_required()
    def __call__(self, info, otherPersonInput, relatedId, relatedTypename):
        mutated_object = CosignerMutationResolver(
            info, otherPersonInput, related_id=relatedId, related_typename=relatedTypename
        ).perform_mutation()
        return mutated_object


class AddCosignerByIdMutationResolver(AdditionManyToManyResolver):
    object_lookup_field = "id"
    m2m_relation_field = "cosigners"

    def get_parent_instance(self):
        qs = get_queryset_by_typename(self.graphql_info.context["request"].user, self.related_typename)
        return qs.get(pk=self.model_lookup_value)

    def get_queryset(self):
        return self.graphql_info.context["request"].user.otherperson_set.all()

    @mutation.field("addCosigner")
    @login_required()
    @assess_state_required()
    def __call__(self, info, relatedId, otherPersonId, relatedTypename, *args, **kwargs):
        obj = AddCosignerByIdMutationResolver(
            info=info, id=relatedId, related_id=otherPersonId, related_typename=relatedTypename
        ).add()
        return obj


class CosignerDeletionResolver(DeletionManyToManyResolver):
    object_lookup_field = "id"
    m2m_relation_field = "cosigners"

    def get_parent_instance(self):
        qs = get_queryset_by_typename(self.graphql_info.context["request"].user, self.related_typename)
        return qs.get(pk=self.related_object_id)

    @mutation.field("deleteCosigner")
    @login_required()
    @assess_state_required()
    def __call__(self, info, cosignerId, relatedId, relatedTypename, *args, **kwargs):
        obj = CosignerDeletionResolver(
            info=info, id=cosignerId, related_id=relatedId, related_typename=relatedTypename
        ).destroy()
        return obj


class CosignerMarkNoResolver(MarkNoManyToManyResolver):
    boolean_field_name = "has_no_cosigners"

    @mutation.field("markNoCosigners")
    @login_required()
    @assess_state_required()
    def __call__(self, info, id, typename, *args, **kwargs):
        obj = CosignerMarkNoResolver(info, id, typename).mutate()
        return obj

[ccsv]

@maxmorlocke

Neat I think something like this would be good to put in, especially if users of the library is writing less code than Strawberry. Maybe it would be useful to lay out something like the class based views in Django to load in mixins for the Query and mutations that are optional for users.

I am willing to add docs if you know what format it needs to be in. I did some of the docs for graphene early on but I don't use github all the time so I might have issues on Pull request.

I guess the layout for docs should be something like this:

• Types (for def)
  ** General stuff (textfield, Character; Int / float; Boolean)
  ** Choices (enum, integerChoice)
  ** Foreign key
  ** Many to Many
  ** Built in Scalars (Datetime, JSON)

• Queries
  ** Filtering
  ** Ordering

• Mutation
  ** Create
  ** Update
  ** Delete
  ** Nested Mutations

• Resolver

• Permissions

[maxmorlocke]

If you want to put together the skeleton of this stuff, I'm happy to share the examples from Lexria's codebase in the docs... just not employed there any longer and don't have the time/will to productionize that for everyone.

[ccsv]

@maxmorlocke Ok I will start working on some .rst files I did a PR for the docs for types it needs some work

[ccsv]

Pull request

#54

[ccsv]

@maxmorlocke ok I added 3 Pull request one for permissions, types, and queries. They are all .rst files I accidentally saved a markdown version as an rst but it is fixed now.

[maxmorlocke]

Thanks, they should all be merged.