161 lines
5.2 KiB
Python
161 lines
5.2 KiB
Python
from flask import jsonify, request
|
|
from api import require_auth
|
|
from api.setting import setting_bp
|
|
from db import get_db
|
|
from db.models.WebhookSetting import WebhookSetting
|
|
|
|
|
|
@setting_bp.route('/path/webhook/', methods=['GET'])
|
|
@require_auth(roles=['admin'])
|
|
def list_webhook_settings():
|
|
"""
|
|
List all webhook settings.
|
|
|
|
This endpoint retrieves a list of all webhook settings in the system.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Returns:
|
|
A JSON array containing all webhook settings.
|
|
|
|
Response Codes:
|
|
- 200: Success
|
|
"""
|
|
with get_db() as session:
|
|
settings = session.query(WebhookSetting).all()
|
|
return jsonify([s.to_dict() for s in settings]), 200
|
|
|
|
|
|
@setting_bp.route('/path/webhook/<int:setting_id>', methods=['GET'])
|
|
@require_auth(roles=['admin'])
|
|
def webhook_setting(setting_id):
|
|
"""
|
|
Get a specific webhook setting by ID.
|
|
|
|
This endpoint retrieves a webhook setting by its ID.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Request:
|
|
- setting_id (int): The ID of the webhook setting to retrieve
|
|
|
|
Returns:
|
|
A JSON object containing the webhook setting.
|
|
|
|
Response Codes:
|
|
- 200: Success
|
|
- 204: No content (setting not found)
|
|
"""
|
|
with get_db() as session:
|
|
setting = session.query(WebhookSetting).filter(WebhookSetting.id == setting_id).first()
|
|
if not setting:
|
|
return jsonify({}), 204
|
|
return jsonify(setting.to_dict()), 200
|
|
|
|
|
|
@setting_bp.route('/path/webhook/', methods=['POST'])
|
|
@require_auth(roles=['admin'])
|
|
def create_webhook_setting():
|
|
"""
|
|
Create a new webhook setting.
|
|
|
|
This endpoint creates a new webhook setting with the provided parameters.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Request:
|
|
- webhook_id (int): The ID of the webhook to associate
|
|
- recursive (bool, optional): Whether the webhook applies to subpaths (default: false)
|
|
- additional_header (str, optional): Additional headers for the webhook (default: '')
|
|
- enabled (bool, optional): Whether the webhook is enabled (default: true)
|
|
- on_events (int, optional): Event types that trigger the webhook (default: 1)
|
|
|
|
Returns:
|
|
A JSON object containing the created webhook setting.
|
|
|
|
Response Codes:
|
|
- 201: Created successfully
|
|
"""
|
|
data = request.json
|
|
with get_db() as session:
|
|
setting = WebhookSetting(
|
|
webhook_id=data.get('webhook_id'),
|
|
recursive=data.get('recursive', False),
|
|
additional_header=data.get('additional_header', ''),
|
|
enabled=data.get('enabled', True),
|
|
on_events=data.get('on_events', 1),
|
|
)
|
|
session.add(setting)
|
|
session.commit()
|
|
return jsonify(setting.to_dict()), 201
|
|
|
|
|
|
@setting_bp.route('/path/webhook/<int:setting_id>', methods=['PUT', 'PATCH'])
|
|
@require_auth(roles=['admin'])
|
|
def update_webhook_setting(setting_id):
|
|
"""
|
|
Update a webhook setting.
|
|
|
|
This endpoint updates an existing webhook setting with the provided parameters.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Request:
|
|
- setting_id (int): The ID of the webhook setting to update
|
|
- webhook_id (int, optional): The new webhook ID
|
|
- recursive (bool, optional): Whether the webhook applies to subpaths
|
|
- additional_header (str, optional): Additional headers for the webhook
|
|
- enabled (bool, optional): Whether the webhook is enabled
|
|
- on_events (int, optional): Event types that trigger the webhook
|
|
|
|
Returns:
|
|
A JSON object containing the updated webhook setting.
|
|
|
|
Response Codes:
|
|
- 200: Updated successfully
|
|
- 404: Webhook setting not found
|
|
"""
|
|
data = request.json
|
|
with get_db() as session:
|
|
setting = session.query(WebhookSetting).get(setting_id)
|
|
if not setting:
|
|
return jsonify({'error': 'Webhook setting not found'}), 404
|
|
if 'webhook_id' in data:
|
|
setting.webhook_id = data['webhook_id']
|
|
if 'recursive' in data:
|
|
setting.recursive = data['recursive']
|
|
if 'additional_header' in data:
|
|
setting.additional_header = data['additional_header']
|
|
if 'enabled' in data:
|
|
setting.enabled = data['enabled']
|
|
if 'on_events' in data:
|
|
setting.on_events = data['on_events']
|
|
|
|
session.commit()
|
|
return jsonify(setting.to_dict()), 200
|
|
|
|
|
|
@setting_bp.route('/path/webhook/<int:setting_id>', methods=['DELETE'])
|
|
@require_auth(roles=['admin'])
|
|
def delete_webhook_setting(setting_id):
|
|
"""
|
|
Delete a webhook setting.
|
|
|
|
This endpoint deletes an existing webhook setting.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Request:
|
|
- setting_id (int): The ID of the webhook setting to delete
|
|
|
|
Returns:
|
|
A JSON object with a success message.
|
|
|
|
Response Codes:
|
|
- 200: Deleted successfully
|
|
- 404: Webhook setting not found
|
|
"""
|
|
with get_db() as session:
|
|
setting = session.query(WebhookSetting).get(setting_id)
|
|
if not setting:
|
|
return jsonify({'error': 'Webhook setting not found'}), 404
|
|
|
|
session.delete(setting)
|
|
session.commit()
|
|
return jsonify({'message': 'Webhook setting deleted'}), 200
|