@staticmethod
def add_issue_webhook(project_key, issue_key):
user_id = session.get('user_id')
if not user_id:
return ('Not authenticated'), 401
user_token = UserToken.query.filter_by(user_id=user_id).first()
if not user_token:
return ('No tokens found for the authenticated user'), 400
access_token = user_token.access_token
headers = {
'Authorization': f'Bearer {access_token}',
'Accept': 'application/json',
'Content-Type': 'application/json'
}
# Get cloud_id
resources_response = requests.get(resources_url, headers=headers)
if resources_response.status_code != 200:
return ('Failed to fetch accessible resources.', resources_response.text), resources_response.status_code
try:
resources = resources_response.json()
except ValueError:
return ('Failed to parse resources response as JSON', resources_response.text), resources_response.status_code
if not resources:
return ('No accessible resources found for the user.'), 404
cloud_id = resources[0].get('id')
if not cloud_id:
return ('Could not determine cloud_id from accessible resources.'), 400
# Validate the project and issue
project_url = f"{base_api_url}/project/{project_key}"
project_response = requests.get(project_url, headers=headers)
if project_response.status_code != 200:
return ('Invalid project key. No project found with the given key'), 404
# Validate the specific issue
issue_url = f"{base_api_url}/issue/{issue_key}"
issue_response = requests.get(issue_url, headers=headers)
if issue_response.status_code != 200:
return ('Invalid issue key. No issue found with the given key in the project'), 404
# Check if a webhook already exists for this target and user
existing_webhook = WebhookInfo.query.filter_by(
user_id=user_id,
target_id=project_key,
target_issue_key=issue_key
).first()
if existing_webhook:
return ('Webhook already exists for the given project and issue'), 400
# Create JQL filter specific to the issue
jql_filter = f'parent = "{issue_key}"'
webhook_url = Config.WEBHOOK_URL
# Prepare the payload
data = {
"url": webhook_url,
"webhooks":[
{
"events": [
"jira:issue_created",
"jira:issue_updated"
],
"jqlFilter": jql_filter,
"enabled": True
}
]
}
create_webhook_url = f"{base_api_url}/webhook"
# Send the request to create the webhook
response = requests.post(create_webhook_url, headers=headers, json=data)
if response.status_code not in [200, 201]:
try:
error_details = response.json()
except ValueError:
error_details = response.text
return ('Failed to create webhook', error_details), response.status_code
try:
response_data = response.json()
except ValueError:
return ('Failed to parse webhook creation response as JSON', response.text), response.status_code
webhook_results = response_data.get('webhookRegistrationResult', [])
print(response_data)
if webhook_results:
webhook_id = webhook_results[0].get('createdWebhookId')
else:
webhook_id = None
if not webhook_id:
return ('No webhook id returned'), 500
# Store webhook information in the database
webhook_info = WebhookInfo(
user_id=user_id,
target_id=project_key,
webhook_id=webhook_id,
target_type='issue',
target_issue_key=issue_key
)
db.session.add(webhook_info)
db.session.commit()
logging.info(f"WebhookInfo added for issue: {webhook_info}")
return ('Webhook created successfully for specific issue', webhook_id), 201
You must be a registered user to add a comment. If you've already registered, sign in. Otherwise, register and sign in.