Implement Deactivate user endpoint on Connect

commcare_connect/connect_id_client/main.py

@@ -9,6 +9,7 @@
     Message,
     MessagingBulkResponse,
     MessagingResponse,
+    UserInfo,
 )
 from commcare_connect.organization.models import Organization
 
@@ -67,6 +68,16 @@ def filter_users(country_code: str, credential: list[str]):
     return [ConnectIdUser(**user_dict) for user_dict in data["found_users"]]
 
 
+def get_user_info(token: str) -> UserInfo:
+    # Returns userinfo for mobile user from ConnectID using access_token.
+    response = httpx.request(
+        GET, f"{settings.CONNECTID_URL}/o/userinfo/", headers={"Authorization": f"Bearer {token}"}
+    )
+    response.raise_for_status()
+    data = response.json()
+    return UserInfo(**data)
+
+
 def _make_request(method, path, params=None, json=None, timeout=5) -> Response:
     if json and not method == "POST":
         raise ValueError("json can only be used with POST requests")

commcare_connect/connect_id_client/models.py

@@ -86,3 +86,14 @@ class Credential:
 
     def __str__(self) -> str:
         return self.name
+
+
+@dataclasses.dataclass
+class UserInfo:
+    sub: str
+    name: str
+    phone: str
+    is_active: bool = True
+
+    def __str__(self) -> str:
+        return self.name

commcare_connect/users/urls.py

@@ -4,6 +4,7 @@
     SMSStatusCallbackView,
     accept_invite,
     create_user_link_view,
+    deactivate_user,
     demo_user_tokens,
     start_learn_app,
     user_detail_view,
@@ -21,4 +22,5 @@
     path("accept_invite/<slug:invite_id>/", view=accept_invite, name="accept_invite"),
     path("demo_users/", view=demo_user_tokens, name="demo_users"),
     path("sms_status_callback/", SMSStatusCallbackView.as_view(), name="sms_status_callback"),
+    path("deactivate_user/", deactivate_user, name="deactivate_user"),
 ]

commcare_connect/users/views.py

@@ -19,7 +19,7 @@
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
-from commcare_connect.connect_id_client.main import fetch_demo_user_tokens
+from commcare_connect.connect_id_client.main import fetch_demo_user_tokens, get_user_info
 from commcare_connect.opportunity.models import Opportunity, OpportunityAccess, UserInvite, UserInviteStatus
 
 from .helpers import create_hq_user
@@ -161,3 +161,19 @@ def post(self, *args, **kwargs):
                 user_invite.status = UserInviteStatus.sms_not_delivered
             user_invite.save()
         return Response(status=200)
+
+
+@csrf_exempt
+@api_view(["POST"])
+@authentication_classes([OAuth2Authentication])
+def deactivate_user(request):
+    username = request.POST.get("username")
+    try:
+        user = User.objects.get(username=username)
+    except User.DoesNotExist:
+        return Response("User does not exists.", status=400)
+    user_info = get_user_info(request.auth.token)
+    if not user_info.is_active:
+        user.is_active = False
+        user.save()
+    return Response(status=200)