[Python] Google watch api integration for real time task create, updated and delete...

{'error': {'code': 403, 'message': "Method doesn't allow unregistered callers (callers without established identity). Please use API Key or other form of API consumer identity to call this API.", 'errors': [{'message': "Method doesn't allow unregistered callers (callers without established identity). Please use API Key or other form of API consumer identity to call this API.", 'domain': 'global', 'reason': 'forbidden'}], 'status': 'PERMISSION_DENIED'}} Internal Server Error: /calendar/google-calendar/webhook/

i am getting above error for the google watch api and push notfication functionality for google calendar, below is my code which i have implemneted

path('sync', AddGoogleCalendarDataAPIView.as_view(), name='sync'),
path('google-calendar/webhook/', GoogleCalendarWebhookAPIView.as_view(), name='google_calendar_webhook'),


class AddGoogleCalendarDataAPIView(CreateAPIView):
"""
API view for adding and updating tasks based on Google Calendar data.
This view handles POST requests to synchronize tasks with data from Google Calendar.
It updates existing tasks or creates new tasks based on the Google Calendar events.
It also removes tasks that no longer exist in the Google Calendar data.

Authentication:
Requires JWT authentication.

Permissions:
Requires the user to be authenticated.

Serializer:
Uses `UpdateTasksCalendarSerializer` for validating and serializing task data.

HTTP Methods:
POST: Updates or creates tasks based on the provided Google Calendar data.

Attributes:
serializer_class (class): The serializer class used for validating and serializing task data.
permission_classes (list): The list of permission classes required for access.
authentication_classes (list): The list of authentication classes used for authentication.

Methods:
get_permissions():
Dynamically assigns permissions based on the RoleID of the user.
create(request, *args, **kwargs):
Handles POST requests to update or create tasks based on Google Calendar data.
"""
serializer_class = UpdateTasksCalendarSerializer
permission_classes = [IsAuthenticated, PageAccessCalendarPermission]
authentication_classes = [JWTAuthentication]

def get_permissions(self):
"""
Dynamically assigns permissions based on the RoleID of the user.
If the user has RoleID = 1, only IsAuthenticated permission is applied.
Otherwise, the default permissions are applied.

Returns:
List of permissions to be applied for the request.
"""
if not self.request.user.is_authenticated:
return [IsAuthenticated()]

if self.request.user.RoleID == 1:
return [IsAuthenticated()]

return [permission() for permission in self.permission_classes]

def create(self, request, *args, **kwargs):
"""
Handles POST requests to update or create tasks based on Google Calendar data.
It fetches the calendar data from the Google Calendar API using the provided access token,
updates or creates tasks based on the data, and deletes tasks that are no longer present in the calendar.

The method behaves differently based on the user's RoleID:
- If RoleID = 1, it skips the agency-related logic.
- If RoleID != 1, it checks the agency condition for task creation and updating.

Args:
request: The HTTP request object.
*args: Variable length argument list.
**kwargs: Arbitrary keyword arguments.

Returns:
Response: A response containing the result of the task synchronization.
"""
try:
calendar_id = request.data.get("calendar_id", "primary")
webhook_url = f"https://f14b-49-43-33-165.ngrok-free.app/calendar/google-calendar/webhook/"
# Start watching Google Calendar for event updates
watch_response = start_google_calendar_watch(self.request.data.get('accessToken'), calendar_id, webhook_url)
print(watch_response)

if "error" in watch_response:
return Response({"error": watch_response}, status=400)

# Check the RoleID to decide if agency-related logic should be executed
user_agent = None
if self.request.user.RoleID != 1:
user_agent = self.request.user.agent_profile.AgencyID if self.request.user.agent_profile else None

meeting_type = self.request.data.get('MeetingType')
if meeting_type is None or meeting_type != "2":
return Response(
{'error': constants.INVALID_GOOGLE_MEETING_ID_MSG},
status=status.HTTP_400_BAD_REQUEST
)

url = constants.GOOGLE_CALENDAR_API_URL
headers = {'Authorization': self.request.data.get('accessToken')}
calendar_api_response = requests.get(url=url, headers=headers)
calendar_api_response.raise_for_status()
calendar_api_response_data = calendar_api_response.json()
items = calendar_api_response_data.get('items', [])

# Modify query to check GoogleCalID condition without agency if RoleID is 1
if self.request.user.RoleID == 1:
tasks_google_cal_id = Tasks.objects.values_list('GoogleCalID', flat=True).filter(
Q(UserID=self.request.user.id) | Q(AssignedTo=self.request.user.id)
).distinct()
else:
tasks_google_cal_id = Tasks.objects.values_list('GoogleCalID', flat=True).filter(
Q(UserID=self.request.user.id) | Q(AssignedTo=self.request.user.id), Agency=user_agent
).distinct()

updated_or_created_tasks = [] # Collect tasks here

for item in items:
if item.get('id') in tasks_google_cal_id:
task = Tasks.objects.filter(
GoogleCalID=item.get('id')
).filter(
Q(UserID=self.request.user.id) | Q(AssignedTo=self.request.user.id)
).distinct().first() # Use `.first()` to get a single task

if task:
task.UpdateAt = timezone.now()
task.TaskType = 5 if item.get('TaskType') is None else item.get('TaskType')
task.MeetingType = self.request.data.get('MeetingType')
task.StartDate = item.get('start').get('dateTime') or item.get('start').get(
'date') + 'T00:00:00Z'
task.EndDate = item.get('end').get('dateTime') or item.get('end').get('date') + 'T00:00:00Z'
task.Title = item.get('summary') or "Untitled"
task.save(
update_fields=['MeetingType', 'StartDate', 'EndDate', 'Title', 'UpdateAt', 'TaskType'])
updated_or_created_tasks.append(task)
else:
task = Tasks.objects.create(
GoogleCalID=item.get('id'),
CreatedAt=timezone.now(),
TaskType=5 if item.get('TaskType') is None else item.get('TaskType'),
MeetingType=self.request.data.get('MeetingType'),
StartDate=item.get('start').get('dateTime') or item.get('start').get('date') + 'T00:00:00Z',
EndDate=item.get('end').get('dateTime') or item.get('end').get('date') + 'T00:00:00Z',
Title=item.get('summary') or "Untitled",
UserID=self.request.user,
Agency=user_agent if self.request.user.RoleID != 1 else None
# Skip Agency condition if RoleID is 1
)
updated_or_created_tasks.append(task)

# Delete tasks that are not in the API response
Tasks.objects.filter(
~Q(GoogleCalID__in={item.get('id') for item in items}) &
~Q(GoogleCalID__isnull=True) &
~Q(GoogleCalID='')
).delete()

# Serialize all tasks
serializer = UpdateTasksCalendarSerializer(updated_or_created_tasks, many=True)
response = response_converter(True, constants.TASK_SYNC_SUCCESS_MSG, serializer.data)
return Response(response)

except Exception as e:
error_message = str(e)
return Response({'error': error_message}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)


class GoogleCalendarWebhookAPIView(APIView):
"""
API View to handle Google Calendar push notifications.
"""
permission_classes = []
authentication_classes = []

def post(self, request, *args, **kwargs):
"""
Handles incoming webhook notifications from Google Calendar.
"""
print(f"Received webhook: {request.headers}")
try:
event_id = request.data.get("id")
action = request.data.get("action")
calendar_id = request.data.get("calendar_id", "primary")

if action == "deleted":
task_exists = Tasks.objects.filter(GoogleCalID=event_id).exists()
if task_exists:
Tasks.objects.filter(GoogleCalID=event_id).delete()
return Response({"message": "Task deleted successfully"})

# Fetch updated event details from Google API
access_token = self.request.data.get('accessToken') # Fetch from user session or database
url = f"https://www.googleapis.com/calendar/v3/calendars/{calendar_id}/events"
headers = {"Authorization": access_token}
event_data = requests.get(url, headers=headers).json()
print(event_data)
# Determine UserID
user_instance = None
if request.user.is_authenticated:
user_instance = request.user # Assign logged-in user
else:
user_id = request.data.get("user_id") # Get UserID from request if available
if user_id:
user_instance = UserModel.objects.filter(id=2).first()

# Check if task exists, then update or create a new one
task, created = Tasks.objects.update_or_create(
GoogleCalID=event_id,
defaults={
"Title": event_data.get("summary", "Untitled"),
"UpdatedAt": timezone.now(),
"TaskType": 5,
"MeetingType": 2,
"UserID": user_instance, # Assign resolved UserID
}
)

message = "Task created successfully" if created else "Task updated successfully"
return Response({"message": message})

except Exception as e:
return Response({"error": str(e)}, status=500)


import uuid

import requests

GOOGLE_WATCH_URL = "https://www.googleapis.com/calendar/v3/calendars/{calendar_id}/events/watch"


def start_google_calendar_watch(access_token, calendar_id, webhook_url):
"""
Starts watching Google Calendar events using the watch API.

Args:
access_token (str): OAuth access token for Google API.
calendar_id (str): ID of the Google Calendar to watch.
webhook_url (str): Webhook endpoint to receive push notifications.

Returns:
dict: Response from Google API or error details.
"""
print("*******************", access_token)
headers = {
"Authorization": "Bearer ya29.a0AeXRPp6ZmAoaqc65Bpw7fBdC6oGf3C10v4M0WlKNGQAOKK2jsBBqhDtzKXLfSiBFw4UI-vtHa9q_1npO54eozEqT9sVtrSquhlgW5mZBmxmxVatsfoECvtsHtSPLxmSQmdvafgSbZLj5Y4bSdi3yUNTfPLIc71iueZZPthISaCgYKAYkSARISFQHGX2MiISJ8cEgaZsuW2cFlkYlzNg0175",
"Content-Type": "application/json"
}
body = {
"id": str(uuid.uuid4()), # Unique identifier for this watch request
"type": "web_hook",
"address": webhook_url, # Webhook URL where notifications will be sent
"params": {"ttl": "86400"} # Time-to-live (max 24h)
}

response = requests.post('https://www.googleapis.com/calendar/v3/calendars/primary/events/watch', headers=headers, json=body)
print(response.json(), "**********************")
return response.json() if response.status_code == 200 else {"error": response.text}


Please help me out to solve this, thanks in advanced

I have tried many solution like changed and verify the token, also check the google developer setting for the token with their calenda scope

Continue reading...