Merge pull request #90 from silverlogic/user-object-type

baseapp-auth[graphql]: add UserObjectType and UserQueries

Author: nossila

.github/workflows/project-workflow.yml

@@ -18,7 +18,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.11", "3.12"]
+        python-version: ["3.12"]
         django-version: ["4.2", "5.0"]
     steps:
       - uses: actions/checkout@v4

baseapp-auth/baseapp_auth/graphql/filters.py

@@ -0,0 +1,21 @@
+import django_filters
+from django.contrib.auth import get_user_model
+from django.db.models import Q
+
+
+class UsersFilter(django_filters.FilterSet):
+    q = django_filters.CharFilter(method="search")
+
+    order_by = django_filters.OrderingFilter(
+        fields=(
+            ("date_joined", "date_joined"),
+            ("last_login", "last_login"),
+        )
+    )
+
+    class Meta:
+        model = get_user_model()
+        fields = ["q", "order_by"]
+
+    def filtesearchr_q(self, queryset, name, value):
+        return queryset.filter(Q(first_name__icontains=value) | Q(last_name__icontains=value))

baseapp-auth/baseapp_auth/graphql/object_types.py

@@ -0,0 +1,133 @@
+import graphene
+from avatar.templatetags.avatar_tags import avatar_url
+from baseapp_core.graphql import DjangoObjectType, File
+from django.apps import apps
+from django.contrib.auth import get_user_model
+from graphene import relay
+
+from .filters import UsersFilter
+from .permissions import PermissionsInterface
+
+User = get_user_model()
+
+interfaces = (relay.Node, PermissionsInterface)
+
+if apps.is_installed("baseapp_notifications"):
+    from baseapp_notifications.graphql.object_types import NotificationsInterface
+
+    interfaces += (NotificationsInterface,)
+
+
+if apps.is_installed("baseapp_pages"):
+    from baseapp_pages.graphql.object_types import MetadataObjectType, PageInterface
+
+    interfaces += (PageInterface,)
+
+
+class AbstractUserObjectType(object):
+    is_authenticated = graphene.Boolean()
+    full_name = graphene.String()
+    short_name = graphene.String()
+
+    avatar = graphene.Field(File, width=graphene.Int(), height=graphene.Int())
+
+    # Make them not required
+    email = graphene.String()
+    phone_number = graphene.String()
+    is_superuser = graphene.Boolean()
+    is_staff = graphene.Boolean()
+    is_email_verified = graphene.Boolean()
+    password_changed_date = graphene.DateTime()
+    new_email = graphene.String()
+    is_new_email_confirmed = graphene.Boolean()
+
+    class Meta:
+        model = User
+        fields = (
+            "pk",
+            "first_name",
+            "last_name",
+            "full_name",
+            "short_name",
+            "email",
+            "phone_number",
+            "is_superuser",
+            "is_staff",
+            "is_active",
+            "is_email_verified",
+            "date_joined",
+            "password_changed_date",
+            "new_email",
+            "is_new_email_confirmed",
+            "is_authenticated",
+            "pages",
+            "comments",
+            "reactions",
+            "last_login",
+        )
+        interfaces = interfaces
+        filterset_class = UsersFilter
+
+    def resolve_avatar(self, info, width, height):
+        return File(url=avatar_url(self, width, height))
+
+    def resolve_metadata(self, info):
+        return MetadataObjectType(
+            meta_title=self.get_full_name(),
+        )
+
+    def resolve_is_authenticated(self, info):
+        return info.context.user.is_authenticated and self.pk == info.context.user.pk
+
+    def resolve_full_name(self, info):
+        return self.get_full_name()
+
+    def resolve_short_name(self, info):
+        return self.get_short_name()
+
+    def resolve_email(self, info):
+        return view_user_private_field(self, info, "email")
+
+    def resolve_phone_number(self, info):
+        return view_user_private_field(self, info, "phone_number")
+
+    def resolve_is_superuser(self, info):
+        return view_user_private_field(self, info, "is_superuser")
+
+    def resolve_is_staff(self, info):
+        return view_user_private_field(self, info, "is_staff")
+
+    def resolve_is_email_verified(self, info):
+        return view_user_private_field(self, info, "is_email_verified")
+
+    def resolve_password_changed_date(self, info):
+        return view_user_private_field(self, info, "password_changed_date")
+
+    def resolve_new_email(self, info):
+        return view_user_private_field(self, info, "new_email")
+
+    def resolve_is_new_email_confirmed(self, info):
+        return view_user_private_field(self, info, "is_new_email_confirmed")
+
+    @classmethod
+    def get_queryset(cls, queryset, info):
+        if info.context.user.is_anonymous:
+            return queryset.filter(is_active=True)
+        return queryset
+
+    @classmethod
+    def get_node(self, info, id):
+        node = super().get_node(info, id)
+        if not info.context.user.has_perm(f"{User._meta.app_label}.view_user", node):
+            return None
+        return node
+
+
+class UserObjectType(AbstractUserObjectType, DjangoObjectType):
+    class Meta(AbstractUserObjectType.Meta):
+        pass
+
+
+def view_user_private_field(user, info, field_name):
+    if info.context.user.has_perm(f"{User._meta.app_label}.view_user_{field_name}", user):
+        return getattr(user, field_name)

baseapp-auth/baseapp_auth/graphql/permissions.py

@@ -1,15 +1,22 @@
 import graphene
+from django.utils.translation import gettext_lazy as _
 from graphene import relay
 
 
 class PermissionsInterface(relay.Node):
-    has_perm = graphene.Boolean(perm=graphene.String(required=True))
+    has_perm = graphene.Boolean(
+        perm=graphene.String(required=True),
+        description=_("Determine if the logged in user has a specific permission for this object."),
+    )
 
     def resolve_has_perm(self, info, perm, **kwargs):
+        # Builds a permission string of the form "<app_label>.<perm>_<model_name>"
         if "." not in perm:
-            # Builds a permission string of the form "<app_label>.<perm>_<model_name>"
             opts = self._meta
-            codename = "%s_%s" % (perm, opts.model_name)
+            if "_" not in perm:
+                codename = "%s_%s" % (perm, opts.model_name)
+            else:
+                codename = perm
             perm = "%s.%s" % (opts.app_label, codename)
 
         return info.context.user.has_perm(perm, self)

baseapp-auth/baseapp_auth/graphql/queries.py

@@ -0,0 +1,31 @@
+from baseapp_core.graphql import Node
+from django.contrib.auth import get_user_model
+from graphene import Field
+from graphene_django.filter import DjangoFilterConnectionField
+
+from .object_types import UserObjectType
+
+User = get_user_model()
+
+
+def get_user_queries(CustomObjectType):
+    class UsersQueries(object):
+        users = DjangoFilterConnectionField(CustomObjectType)
+        user = Node.Field(CustomObjectType)
+
+        me = Field(CustomObjectType)
+
+        def resolve_users(self, info, **kwargs):
+            if info.context.user.has_perm(f"{User._meta.app_label}.view_all_users"):
+                return CustomObjectType._meta.model.objects.all()
+            return CustomObjectType._meta.model.objects.none()
+
+        def resolve_me(self, info, **kwargs):
+            if info.context.user.is_authenticated:
+                return info.context.user
+            return None
+
+    return UsersQueries
+
+
+UsersQueries = get_user_queries(UserObjectType)

baseapp-auth/baseapp_auth/models.py

@@ -11,7 +11,16 @@
 from .managers import UserManager
 
 
-class AbstractUser(PermissionsMixin, AbstractBaseUser):
+def use_relay_model():
+    try:
+        from baseapp_core.graphql.models import RelayModel
+
+        return RelayModel
+    except ImportError:
+        return object
+
+
+class AbstractUser(PermissionsMixin, AbstractBaseUser, use_relay_model()):
     email = CaseInsensitiveEmailField(unique=True, db_index=True)
     is_email_verified = models.BooleanField(default=False)
     date_joined = models.DateTimeField(_("date joined"), default=timezone.now)
@@ -54,6 +63,17 @@ class Meta:
         abstract = True
         verbose_name = _("user")
         verbose_name_plural = _("users")
+        permissions = [
+            ("view_all_users", _("can view all users")),
+            ("view_user_email", _("can view user's email field")),
+            ("view_user_phone_number", _("can view user's phone number field")),
+            ("view_user_is_superuser", _("can view user's is_superuser field")),
+            ("view_user_is_staff", _("can view user's is_staff field")),
+            ("view_user_is_email_verified", _("can view user's is_email_verified field")),
+            ("view_user_password_changed_date", _("can view user's password_changed_date field")),
+            ("view_user_new_email", _("can view user's new_email field")),
+            ("view_user_is_new_email_confirmed", _("can view user's is_new_email_confirmed field")),
+        ]
 
     def __str__(self):
         return self.get_full_name()

baseapp-auth/baseapp_auth/permissions.py

@@ -0,0 +1,37 @@
+from django.contrib.auth import get_user_model
+from django.contrib.auth.backends import BaseBackend
+
+User = get_user_model()
+
+private_field_perms = [
+    f"{User._meta.app_label}.view_user_email",
+    f"{User._meta.app_label}.view_user_phone_number",
+    f"{User._meta.app_label}.view_user_is_superuser",
+    f"{User._meta.app_label}.view_user_is_staff",
+    f"{User._meta.app_label}.view_user_is_email_verified",
+    f"{User._meta.app_label}.view_user_password_changed_date",
+    f"{User._meta.app_label}.view_user_new_email",
+    f"{User._meta.app_label}.view_user_is_new_email_confirmed",
+]
+
+
+class UsersPermissionsBackend(BaseBackend):
+    def has_perm(self, user_obj, perm, obj=None):
+        if perm == f"{User._meta.app_label}.view_user":
+            # every body can view all users
+            # TO DO: maybe check if user is_active
+            return True
+
+        if perm == f"{User._meta.app_label}.view_all_users":
+            return True
+
+        if perm in private_field_perms and obj is not None:
+            if (
+                isinstance(obj, User)
+                and user_obj.is_authenticated
+                and (obj.pk == user_obj.pk or (user_obj.is_superuser or user_obj.is_staff))
+            ):
+                return True
+            else:
+                # Anyone with permission set also can:
+                return user_obj.has_perm(perm)

baseapp-auth/baseapp_auth/tests/graphql/__init__.py

@@ -0,0 +1,2 @@
+from baseapp_core.graphql.testing.fixtures import *  # noqa
+from baseapp_core.tests.fixtures import *  # noqa

baseapp-auth/baseapp_auth/tests/graphql/conftest.py

@@ -0,0 +1,27 @@
+import pytest
+
+pytestmark = pytest.mark.django_db
+
+QUERY = """
+    query {
+        me {
+            id
+            isAuthenticated
+        }
+    }
+"""
+
+
+def test_anon_cant_query_me(graphql_client):
+    response = graphql_client(QUERY)
+    content = response.json()
+
+    assert content["data"]["me"] is None
+
+
+def test_user_cant_query_me(django_user_client, graphql_user_client):
+    response = graphql_user_client(QUERY)
+    content = response.json()
+
+    assert content["data"]["me"]["id"] == django_user_client.user.relay_id
+    assert content["data"]["me"]["isAuthenticated"] is True