Zendesk to DagKnows integration

import requests from requests.auth import HTTPBasicAuth # Function to create a webhook in Zendesk def list_webhook(email, api_token, subdomain): # Define the endpoint URL for creating webhooks url = f'https://{subdomain}.zendesk.com/api/v2/webhooks' # Define the headers including the authorization headers = { 'Content-Type': 'application/json' } # Make the request to create the webhook response = requests.get(url, auth = HTTPBasicAuth(email + '/token', api_token), headers=headers) # Check if the request was successful if response.status_code == 200: return {'success': True, 'webhooks': response.json()} else: print(response.status_code) return {'success': False, 'error': response.text} creds = _get_creds(zendesk_cred_label)['creds'] api_token = creds['password'] email = creds['username'] op = list_webhook(email, api_token, subdomain) print("Response: ", json.dumps(op, indent=4))

import requests from requests.auth import HTTPBasicAuth # Function to create a webhook in Zendesk def create_webhook(email, api_token, subdomain, webhook_url, dagknows_token): # Define the endpoint URL for creating webhooks url = f'https://{subdomain}.zendesk.com/api/v2/webhooks' # Define the headers including the authorization headers = { 'Content-Type': 'application/json' } # Define the webhook payload payload = { 'webhook': { 'name': 'Keyword Triggered Webhook', 'endpoint': webhook_url, 'http_method': 'POST', 'request_format': 'json', 'status': 'active', 'subscriptions': [ "conditional_ticket_events" ], "authentication": { "type": "bearer_token", "data": { "token": dagknows_token }, "add_position": "header" } } } # Make the request to create the webhook response = requests.post(url, auth = HTTPBasicAuth(email + '/token', api_token), headers=headers, json=payload) # Check if the request was successful if response.status_code == 201: return {'success': True, 'webhook_id': response.json()['webhook']['id']} else: print(response.status_code) return {'success': False, 'error': response.text} creds = _get_creds(zendesk_cred_label)['creds'] api_token = creds['password'] email = creds['username'] dkcreds = _get_creds(dagknows_cred_label)['creds'] dagknows_token = dkcreds['password'] op = create_webhook(email, api_token, subdomain, webhook_url, dagknows_token) print("Response: ", op) webhook_id = "" if op['success']: webhook_id = op['webhook_id']

import requests from requests.auth import HTTPBasicAuth # Function to add a trigger to Zendesk def add_zendesk_trigger(api_url, email, api_token, keyword, webhook_id): headers = { 'Content-Type': 'application/json' } data_to_send = """ { "id" : "{{ticket.id}}", "subject" : "{{ticket.subject}}", "latest_comment" : "{{ticket.latest_comment}}" } """ data = { 'trigger': { 'title': f'Trigger for keyword {keyword}', 'conditions': { 'all': [ { 'field': 'comment_includes_word', 'operator': 'includes', 'value': keyword }, { 'field': 'comment_is_public', 'value': True } ] }, 'actions': [ { 'field': 'notification_webhook', 'value': [webhook_id, data_to_send] } ] } } basic_auth = HTTPBasicAuth(email + '/token', api_token) response = requests.post(api_url, auth = basic_auth, headers=headers, json=data) return response.status_code, response.json() api_url = f'https://{subdomain}.zendesk.com/api/v2/triggers' creds = _get_creds(zendesk_cred_label)['creds'] api_token = creds['password'] email = creds['username'] status_code, response_json = add_zendesk_trigger(api_url, email, api_token, keyword, webhook_id) print(status_code) print(json.dumps(response_json, indent=4))

import requests from requests.auth import HTTPBasicAuth # Function to add a trigger to Zendesk def add_zendesk_trigger(api_url, email, api_token, keyword, webhook_id): headers = { 'Content-Type': 'application/json' } data_to_send = """ { "id" : "{{ticket.id}}", "subject" : "{{ticket.subject}}", "latest_comment" : "{{ticket.description}}" } """ data = { 'trigger': { 'title': f'Trigger for keyword {keyword}', 'conditions': { 'all': [ {'field': 'subject', 'operator': 'contains', 'value': keyword} ] }, 'actions': [ { 'field': 'notification_webhook', 'value': [webhook_id, data_to_send] } ] } } basic_auth = HTTPBasicAuth(email + '/token', api_token) response = requests.post(api_url, auth = basic_auth, headers=headers, json=data) return response.status_code, response.json() api_url = f'https://{subdomain}.zendesk.com/api/v2/triggers' creds = _get_creds(zendesk_cred_label)['creds'] api_token = creds['password'] email = creds['username'] status_code, response_json = add_zendesk_trigger(api_url, email, api_token, keyword, webhook_id) print(status_code) print(json.dumps(response_json, indent=4))

import requests import json from requests.auth import HTTPBasicAuth # Function to add a trigger to Zendesk def add_zendesk_trigger_for_new_ticket(api_url, email, api_token, webhook_id): headers = { 'Content-Type': 'application/json' } data_to_send = """ { "id" : "{{ticket.id}}", "subject" : "{{ticket.subject}}", "description" : "{{ticket.description}}", "created_at" : "{{ticket.created_at}}", "updated_at" : "{{ticket.updated_at}}", "priority" : "{{ticket.priority}}", "status" : "{{ticket.status}}", "assignee_id" : "{{ticket.assignee_id}}", "requester_id" : "{{ticket.requester_id}}", "organization_id" : "{{ticket.organization_id}}" } """ data = { 'trigger': { 'title': 'Trigger for new ticket creation', 'conditions': { 'all': [ { 'field': 'status', 'operator': 'is', 'value': 'new' } ] }, 'actions': [ { 'field': 'notification_webhook', 'value': [webhook_id, data_to_send] } ] } } basic_auth = HTTPBasicAuth(f'{email}/token', api_token) response = requests.post(api_url, auth=basic_auth, headers=headers, json=data) return response.status_code, response.json() # Usage creds = _get_creds(zendesk_cred_label)['creds'] api_token = creds['password'] email = creds['username'] api_url = f'https://{subdomain}.zendesk.com/api/v2/triggers.json' status_code, response_json = add_zendesk_trigger_for_new_ticket(api_url, email, api_token, webhook_id) print(status_code) print(json.dumps(response_json, indent=4))

import requests import json from requests.auth import HTTPBasicAuth # Usage creds = _get_creds(zendesk_cred_label)['creds'] api_token = creds['password'] email = creds['username'] # Function to get group ID by name def get_group_id(subdomain, email, api_token, group_name): api_url = f'https://{subdomain}.zendesk.com/api/v2/groups.json' headers = { 'Content-Type': 'application/json' } basic_auth = HTTPBasicAuth(f'{email}/token', api_token) response = requests.get(api_url, auth=basic_auth, headers=headers) if response.status_code == 200: groups = response.json().get('groups', []) for group in groups: if group['name'].lower() == group_name.lower(): return group['id'] return None # Function to add a trigger to Zendesk def add_zendesk_trigger_for_new_ticket(api_url, email, api_token, webhook_id, group_id): headers = { 'Content-Type': 'application/json' } data_to_send = """ { "id" : "{{ticket.id}}", "subject" : "{{ticket.subject}}", "description" : "{{ticket.description}}", "created_at" : "{{ticket.created_at}}", "updated_at" : "{{ticket.updated_at}}", "priority" : "{{ticket.priority}}", "status" : "{{ticket.status}}", "assignee_id" : "{{ticket.assignee_id}}", "requester_id" : "{{ticket.requester_id}}", "organization_id" : "{{ticket.organization_id}}" } """ data = { 'trigger': { 'title': 'Trigger for new or open ticket creation in DevOps group', 'conditions': { 'all': [ { 'field': 'group_id', 'operator': 'is', 'value': str(group_id) } ], 'any': [ { 'field': 'status', 'operator': 'is', 'value': 'new' }, { 'field': 'status', 'operator': 'is', 'value': 'open' } ] }, 'actions': [ { 'field': 'notification_webhook', 'value': [webhook_id, data_to_send] } ] } } basic_auth = HTTPBasicAuth(f'{email}/token', api_token) response = requests.post(api_url, auth=basic_auth, headers=headers, json=data) return response.status_code, response.json() # Get the group ID for "DevOps" group_id = get_group_id(subdomain, email, api_token, 'DevOps') if group_id: api_url = f'https://{subdomain}.zendesk.com/api/v2/triggers.json' status_code, response_json = add_zendesk_trigger_for_new_ticket(api_url, email, api_token, webhook_id, group_id) print(status_code) print(json.dumps(response_json, indent=4)) else: print("Group 'DevOps' not found.")