graphql_schema.py

import os
import graphene
import hashlib

from django.contrib.auth import get_user_model
from django.db import connection
from django.utils import timezone
from graphql.error import GraphQLError
from graphene_django.types import DjangoObjectType
from graphene_django.views import GraphQLView
from graphene_django.forms.mutation import DjangoModelFormMutation

from portal.models import DataFile, DataSet, Lab, Profile, ProfileToken, ShareGroup
from portal.views import TokenLoginMixin
from portal.forms import DataSetForm


# Protect graphql API page
class PrivateGraphQLView(TokenLoginMixin, GraphQLView):
    pass


# Create a GraphQL type for the various models
class DataFileType(DjangoObjectType):
    class Meta:
        model = DataFile
        fields = "__all__"

    created_by_user = graphene.Boolean()

    def resolve_created_by_user(self, info):
        return self.uploaded_by == info.context.user.profile


class DataSetType(DjangoObjectType):
    class Meta:
        model = DataSet
        fields = "__all__"

    created_by_user = graphene.Boolean()

    def resolve_created_by_user(self, info):
        return self.created_by == info.context.user.profile


class ProfileType(DjangoObjectType):
    class Meta:
        model = Profile
        fields = ['id', 'accountname', 'unix_username', 'user', 'labs']


class ProfileTokenType(DjangoObjectType):
    class Meta:
        model = ProfileToken
        fields = ['id', 'user', 'api_token']


class UserType(DjangoObjectType):
    class Meta:
        model = get_user_model()
        fields = ['id', 'username', 'first_name', 'last_name', 'email', 'is_staff']


class LabType(DjangoObjectType):
    class Meta:
        model = Lab
        fields = ['id', 'name', 'pi', 'data_managers', 'institution']


class ShareGroupType(DjangoObjectType):
    class Meta:
        model = ShareGroup
        fields = ['id', 'name', 'profiles']


# Create a Query type
class Query(graphene.ObjectType):
    datafile = graphene.Field(DataFileType, id=graphene.String(), dbid=graphene.ID())
    dataset = graphene.Field(DataSetType, id=graphene.String(), dbid=graphene.ID())
    datafiles = graphene.List(
        DataFileType,
        key=graphene.String(),
        value=graphene.String(),
        key_values=graphene.JSONString(),
        keys=graphene.List(graphene.String),
        values=graphene.List(graphene.String),
        dataset=graphene.String(),
        file_hash=graphene.String(),
        in_id_list=graphene.List(graphene.String)
    )
    datasets = graphene.List(DataSetType, in_id_list=graphene.List(graphene.String))
    lab = graphene.Field(
        LabType,
        name=graphene.String(),
        dbid=graphene.ID()
    )
    labs = graphene.List(
        LabType,
        in_name_list=graphene.List(graphene.String)
    )
    my_profile = graphene.Field(
        ProfileType
    )
    profiles = graphene.List(
        ProfileType,
        in_email_list=graphene.List(graphene.String)
    )
    profiles_token = graphene.List(
        ProfileTokenType,
        in_email_list=graphene.List(graphene.String)
    )
    sharegroups = graphene.List(
        ShareGroupType,
        in_name_list=graphene.List(graphene.String)
    )

    def resolve_datafile(self, info, **kwargs):
        id = kwargs.get('id')
        dbid = kwargs.get('dbid')
        qs = DataFile.objects.accessible_to_profile(info.context.user.profile)

        if id is not None:
            return qs.filter(iric_data_id=id).first()
        elif dbid is not None:
            return qs.filter(pk=dbid).first()

    def resolve_dataset(self, info, **kwargs):
        id = kwargs.get('id')
        dbid = kwargs.get('dbid')
        qs = DataSet.objects.accessible_to_profile(info.context.user.profile)

        if id is not None:
            return qs.filter(iric_data_id=id).first()
        elif dbid is not None:
            return qs.filter(pk=dbid).first()

        return None

    def resolve_datafiles(self, info, **kwargs):
        qs = DataFile.objects.accessible_to_profile(info.context.user.profile)

        key_values = kwargs.get('key_values', None)
        keys = kwargs.get('keys', None)
        values = kwargs.get('values', None)
        dataset = kwargs.get('dataset', None)
        file_hash = kwargs.get('file_hash', None)
        id_list = kwargs.get('in_id_list', None)

        if key_values is not None:
            for k, v in key_values.items():
                qs = qs.filter(annotations__contains={k: v})

        if keys is not None:
            for k in keys:
                qs = qs.filter(annotations__has_key=k)

        if values is not None:
            for v in values:
                qs = qs.filter(annotations__values__icontains=v)

        if dataset is not None:
            qs = qs.filter(datasets__iric_data_id=dataset)

        if file_hash is not None:
            iric_data_file_hash = os.path.join(file_hash[:3], file_hash)
            qs = qs.filter(file=iric_data_file_hash)

        if id_list is not None:
            qs = qs.filter(iric_data_id__in=id_list)

        return qs.distinct()

    def resolve_datasets(self, info, **kwargs):
        id_list = kwargs.get('in_id_list')
        qs = DataSet.objects.accessible_to_profile(info.context.user.profile)

        if id_list is not None:
            qs =  qs.filter(iric_data_id__in=id_list)

        return qs.distinct()

    def resolve_lab(self, info, **kwargs):
        name = kwargs.get('name')
        dbid = kwargs.get('dbid')

        if name is not None:
            return Lab.objects.get(name=name)
        elif dbid is not None:
            return Lab.objects.get(pk=dbid)

    def resolve_labs(self, info, **kwargs):
        name_list = kwargs.get('in_name_list')
        qs = Lab.objects.all()

        if name_list is not None:
            qs = qs.filter(name__in=name_list)

        return qs.distinct()

    def resolve_my_profile(self, info, **kwargs):
        return Profile.objects.get(user=info.context.user)

    def resolve_profiles(self, info, **kwargs):
        email_list = kwargs.get('in_email_list')
        qs = Profile.objects.all()

        if email_list is not None:
            return qs.filter(user__email__in=email_list)

        return qs.distinct()

    def resolve_profiles_token(self, info, **kwargs):
        email_list = kwargs.get('in_email_list')
        qs = ProfileToken.objects.none()

        if info.context.user.is_staff:
            qs = ProfileToken.objects.all()
            if email_list:
                qs = qs.filter(user__email__in=email_list)
        return qs.distinct()

    def resolve_sharegroups(self, info, **kwargs):
        name_list = kwargs.get('in_name_list')
        qs = ShareGroup.objects.all()

        if name_list is not None:
            qs = qs.filter(name__in=name_list)

        return qs.distinct()


class DataSetMutation(DjangoModelFormMutation):
    """Enable the creation or modification of a DataSet
    Makes use of the DataSetForm, so all exposed fields from that form are available"""

    class Meta:
        form_class = DataSetForm
        return_field_name = 'dataset'

    @classmethod
    def get_form_kwargs(cls, root, info, **input) -> dict:
        kwargs = super().get_form_kwargs(root, info, **input)
        kwargs['request_profile'] = info.context.user.profile
        if 'read_only' not in kwargs['data'] and 'instance' not in kwargs.keys():
            kwargs['data']['read_only'] = True
        return kwargs

    @classmethod
    def perform_mutate(cls, form, info):
        object = form.save(commit=False)
        if not object.pk:
            object.created_by = info.context.user.profile

        object.last_update_by = info.context.user.profile
        object.save()
        return super().perform_mutate(form, info)


class DataFileAttachMutation(graphene.Mutation):
    attached_datafile = graphene.Field(DataFileType)

    class Arguments:
        # The input arguments for this mutation
        file_hash = graphene.String(required=True)
        filename = graphene.String(required=True)
        target_user_email = graphene.String(required=True)
        annotations = graphene.JSONString()

    @classmethod
    def mutate(cls, root, info, file_hash, filename, target_user_email, annotations=None):
        # check that calling user is staff:
        if not info.context.user.is_staff:
            raise GraphQLError("You are not allowed to use this API function")

        # transform file_hash
        iric_data_file_hash = os.path.join(file_hash[:3], file_hash)

        target_profile = Profile.objects.get(user__email=target_user_email)

        if not DataFile.objects.filter(file=iric_data_file_hash).exists():
            raise GraphQLError("No datafile associated with the provided hash")

        with connection.cursor() as cursor:
            cursor.execute(
                "INSERT INTO portal_datafile(file, filename, uploaded_by_id, upload_timestamp, annotations) VALUES (%s, %s, %s, %s, %s)",
                [
                    iric_data_file_hash,
                    filename,
                    target_profile.id,
                    timezone.now(),
                    annotations
                ]
            )

            df = DataFile.objects.latest('id')

        df.iric_data_id = "DF{}".format(hashlib.md5(str(df.id).encode()).hexdigest()[:8]).upper()
        df.save()

        return DataFileAttachMutation(attached_datafile=df)


class Mutation(graphene.ObjectType):
    dataset = DataSetMutation.Field()
    attach_datafile_by_hash = DataFileAttachMutation.Field()


schema = graphene.Schema(query=Query, mutation=Mutation, auto_camelcase=False)