webhook-add

services/auth.py

@@ -4,32 +4,11 @@ import httpx
 
 from cache.cache import get_cached_author_by_user_id
 from resolvers.stat import get_with_stat
-from settings import ADMIN_SECRET, AUTH_URL
+from services.schema import request_graphql_data
+from settings import ADMIN_SECRET
 from utils.logger import root_logger as logger
 
 
-async def request_data(gql, headers=None):
-    if headers is None:
-        headers = {"Content-Type": "application/json"}
-    try:
-        # logger.debug(f"{AUTH_URL} {gql} {headers}")
-        async with httpx.AsyncClient() as client:
-            response = await client.post(AUTH_URL, json=gql, headers=headers)
-            if response.status_code == 200:
-                data = response.json()
-                errors = data.get("errors")
-                if errors:
-                    logger.error(f"{AUTH_URL} response: {data}")
-                else:
-                    return data
-    except Exception as _e:
-        # Handling and logging exceptions during authentication check
-        import traceback
-
-        logger.error(f"request_data error: {traceback.format_exc()}")
-    return None
-
-
 async def check_auth(req):
     token = req.headers.get("Authorization")
     user_id = ""
@@ -48,7 +27,7 @@ async def check_auth(req):
             "variables": variables,
             "operationName": operation,
         }
-        data = await request_data(gql)
+        data = await request_graphql_data(gql)
         if data:
             logger.debug(data)
             user_data = data.get("data", {}).get(query_name, {}).get("claims", {})
@@ -71,7 +50,7 @@ async def add_user_role(user_id):
         "variables": variables,
         "operationName": operation,
     }
-    data = await request_data(gql, headers)
+    data = await request_graphql_data(gql, headers=headers)
     if data:
         user_id = data.get("data", {}).get(query_name, {}).get("id")
         return user_id

services/schema.py

@@ -1,5 +1,31 @@
+from asyncio.log import logger
 from ariadne import MutationType, QueryType
+import httpx
+
+from settings import AUTH_URL
 
 query = QueryType()
 mutation = MutationType()
 resolvers = [query, mutation]
+
+
+async def request_graphql_data(gql, url=AUTH_URL, headers=None):
+    if headers is None:
+        headers = {"Content-Type": "application/json"}
+    try:
+        # logger.debug(f"{url} {gql} {headers}")
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, json=gql, headers=headers)
+            if response.status_code == 200:
+                data = response.json()
+                errors = data.get("errors")
+                if errors:
+                    logger.error(f"{url} response: {data}")
+                else:
+                    return data
+    except Exception as _e:
+        # Handling and logging exceptions during authentication check
+        import traceback
+
+        logger.error(f"request_graphql_data error: {traceback.format_exc()}")
+    return None

services/webhook.py

@@ -1,4 +1,5 @@
 import asyncio
+from asyncio.log import logger
 import os
 import re
 
@@ -12,6 +13,37 @@ from cache.cache import cache_author
 from orm.author import Author
 from resolvers.stat import get_with_stat
 from services.db import local_session
+from services.schema import request_graphql_data
+from settings import ADMIN_SECRET
+
+
+async def create_webhook_endpoint():
+    # TODO: add webhook to authorizer with graphql request
+
+    headers = {
+        "Content-Type": "application/json",
+        "x-authorizer-admin-secret": ADMIN_SECRET,
+    }
+
+    # https://docs.authorizer.dev/core/graphql-api#_add_webhook
+    operation = "AddWebhook"
+    query_name = "_add_webhook"
+    variables = {
+        "params": {
+            "event_name": "user.login",
+            "endpoint": "https://core.dscrs.site/new-author",
+            "enabled": True,
+        }
+    }
+    gql = {
+        "query": f"query {operation}($params: AddWebhookRequest!)  {{"
+        + f"{query_name}(params: $params) {{ message }} "
+        + "}",
+        "variables": variables,
+        "operationName": operation,
+    }
+    result = await request_graphql_data(gql, headers=headers)
+    logger.info(result)
 
 
 class WebhookEndpoint(HTTPEndpoint):