Zato provides two ways to integrate with Jira:
self.cloud.jiraUse JQL (Jira Query Language) to search for issues. The connector returns structured data you can process in Python.
# -*- coding: utf-8 -*-
from zato.server.service import Service
class GetOpenTickets(Service):
name = 'my.jira.get-open-tickets'
def handle(self):
# Name of your Jira connection defined in Dashboard
conn_name = 'My Jira'
# Fields to retrieve
fields = ['key', 'summary', 'status', 'assignee', 'created']
# Get a reference to the Jira connection
jira = self.cloud.jira[conn_name]
# Obtain a client
with jira.conn.client() as client:
# Build and execute JQL query
query = 'status="Open" AND project="MYPROJECT"'
result = client.jql(jql=query, fields=fields)
# Process results
issues = result.get('issues', [])
for issue in issues:
key = issue['key']
summary = issue['fields']['summary']
self.logger.info(f'Found issue: {key} - {summary}')
self.response.payload = {'count': len(issues)}
Move issues through workflow states by specifying the transition ID.
# -*- coding: utf-8 -*-
import os
from json import dumps
import requests
from zato.server.service import Service
class TransitionIssue(Service):
name = 'my.jira.transition-issue'
input = 'ticket_id', 'transition_id'
def handle(self):
# Jira credentials
base_url = 'https://yourcompany.atlassian.net'
username = 'your.email@company.com'
password = os.environ['JIRA_API_TOKEN']
auth = (username, password)
# Build the transition request
url = f'{base_url}/rest/api/2/issue/{self.request.input.ticket_id}/transitions'
data = {'transition': {'id': self.request.input.transition_id}}
headers = {'Content-Type': 'application/json'}
# Execute the transition
response = requests.post(url, data=dumps(data), auth=auth, headers=headers)
if not str(response.status_code).startswith('2'):
raise Exception(f'Transition failed: {response.status_code} {response.text}')
self.response.payload = {'status': 'transitioned'}
Post comments to issues, with control over visibility (internal vs customer-visible in Service Desk).
# -*- coding: utf-8 -*-
import os
from json import dumps
import requests
from zato.server.service import Service
class AddComment(Service):
name = 'my.jira.add-comment'
input = 'ticket_id', 'comment', '-public'
def handle(self):
# Jira credentials
base_url = 'https://yourcompany.atlassian.net'
username = 'your.email@company.com'
password = os.environ['JIRA_API_TOKEN']
auth = (username, password)
# For Service Desk, use the service desk API
url = f'{base_url}/rest/servicedeskapi/request/{self.request.input.ticket_id}/comment'
data = {
'body': self.request.input.comment,
'public': self.request.input.public or False
}
headers = {'Content-Type': 'application/json'}
response = requests.post(url, data=dumps(data), auth=auth, headers=headers)
if not str(response.status_code).startswith('2'):
raise Exception(f'Comment failed: {response.status_code} {response.text}')
self.response.payload = {'status': 'commented'}
Download attachments from tickets. Jira returns a redirect URL for the actual file content.
# -*- coding: utf-8 -*-
import os
import requests
from zato.server.service import Service
class GetAttachment(Service):
name = 'my.jira.get-attachment'
input = 'attachment_id'
def handle(self):
# Jira credentials
base_url = 'https://yourcompany.atlassian.net'
username = 'your.email@company.com'
password = os.environ['JIRA_API_TOKEN']
auth = (username, password)
# Get attachment metadata and download URL
url = f'{base_url}/rest/api/3/attachment/content/{self.request.input.attachment_id}'
# Don't follow redirects - we want the Location header
response = requests.get(url, auth=auth, allow_redirects=False)
# The actual file is at the redirect location
download_url = response.headers['Location']
# Now download the file
file_response = requests.get(download_url)
self.response.payload = {
'content': file_response.content,
'size': len(file_response.content)
}
The Jira cloud connector simplifies common operations. Configure it in Dashboard under Connections → Cloud → Jira.
# -*- coding: utf-8 -*-
from zato.server.service import Service
class JiraConnectorExample(Service):
name = 'my.jira.connector-example'
def handle(self):
# Get the Jira connection
jira = self.cloud.jira['My Jira']
with jira.conn.client() as client:
# Get a single issue
issue = client.get_issue('PROJ-123', fields=['summary', 'status'])
# Search with JQL
results = client.jql(
jql='project=PROJ AND status=Open',
fields=['key', 'summary']
)
# The client provides direct access to Jira's API
self.response.payload = {
'issue': issue,
'search_count': len(results.get('issues', []))
}