131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
from flask import Blueprint, jsonify, request
|
|
from api import require_auth
|
|
from db import get_db
|
|
from db.models.Webhook import Webhook
|
|
|
|
webhook_bp = Blueprint('webhook', __name__, url_prefix='/api/webhook')
|
|
|
|
|
|
@webhook_bp.route('/', methods=['GET'])
|
|
@require_auth(roles=['admin'])
|
|
def list_webhooks():
|
|
"""
|
|
List all webhooks.
|
|
|
|
This endpoint retrieves a list of all webhooks in the system.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Returns:
|
|
A JSON array containing all webhooks.
|
|
|
|
Response Codes:
|
|
- 200: Success
|
|
"""
|
|
with get_db() as session:
|
|
hooks = session.query(Webhook).all()
|
|
return jsonify([h.to_dict() for h in hooks]), 200
|
|
|
|
|
|
@webhook_bp.route('/', methods=['POST'])
|
|
@require_auth(['admin'])
|
|
def create_webhook():
|
|
"""
|
|
Create a new webhook.
|
|
|
|
This endpoint creates a new webhook with the provided URL.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Request:
|
|
- hook_url (str): The URL of the webhook
|
|
|
|
Returns:
|
|
A JSON object containing the created webhook.
|
|
|
|
Response Codes:
|
|
- 201: Created successfully
|
|
- 400: Bad request (hook_url is missing)
|
|
- 409: Conflict (webhook URL already exists)
|
|
"""
|
|
data = request.json
|
|
hook_url = data.get('hook_url')
|
|
if not hook_url:
|
|
return jsonify({'error': 'hook_url required'}), 400
|
|
|
|
with get_db() as session:
|
|
existing = session.query(Webhook).filter_by(hook_url=hook_url).first()
|
|
if existing:
|
|
return jsonify({'error': 'Webhook URL already exists'}), 409
|
|
webhook = Webhook(hook_url=hook_url)
|
|
session.add(webhook)
|
|
session.commit()
|
|
return jsonify(webhook.to_dict()), 201
|
|
|
|
|
|
@webhook_bp.route('/<int:webhook_id>', methods=['PUT', 'PATCH'])
|
|
@require_auth(roles=['admin'])
|
|
def update_webhook(webhook_id):
|
|
"""
|
|
Update a webhook.
|
|
|
|
This endpoint updates an existing webhook with the provided URL.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Request:
|
|
- webhook_id (int): The ID of the webhook to update
|
|
- hook_url (str): The new URL for the webhook
|
|
|
|
Returns:
|
|
A JSON object containing the updated webhook.
|
|
|
|
Response Codes:
|
|
- 200: Updated successfully
|
|
- 400: Bad request (hook_url is missing)
|
|
- 404: Webhook not found
|
|
- 409: Conflict (webhook URL already exists)
|
|
"""
|
|
data = request.json
|
|
if 'hook_url' not in data:
|
|
return jsonify({'error': 'hook_url is required'}), 400
|
|
|
|
with get_db() as session:
|
|
webhook = session.query(Webhook).get(webhook_id)
|
|
if not webhook:
|
|
return jsonify({'error': 'Webhook not found'}), 404
|
|
|
|
existing = session.query(Webhook).filter_by(hook_url=data['hook_url']).filter(Webhook.id != webhook_id).first()
|
|
if existing:
|
|
return jsonify({'error': 'Webhook URL already exists'}), 409
|
|
|
|
webhook.hook_url = data['hook_url']
|
|
session.commit()
|
|
return jsonify(webhook.to_dict()), 200
|
|
|
|
|
|
@webhook_bp.route('/<int:webhook_id>', methods=['DELETE'])
|
|
@require_auth(roles=['admin'])
|
|
def delete_webhook(webhook_id):
|
|
"""
|
|
Delete a webhook.
|
|
|
|
This endpoint deletes an existing webhook.
|
|
It requires authentication with the 'admin' role.
|
|
|
|
Request:
|
|
- webhook_id (int): The ID of the webhook to delete
|
|
|
|
Returns:
|
|
A JSON object with a success message.
|
|
|
|
Response Codes:
|
|
- 200: Deleted successfully
|
|
- 404: Webhook not found
|
|
"""
|
|
with get_db() as session:
|
|
webhook = session.query(Webhook).get(webhook_id)
|
|
if not webhook:
|
|
return jsonify({'error': 'Webhook not found'}), 404
|
|
|
|
session.delete(webhook)
|
|
session.commit()
|
|
return jsonify({'message': 'Webhook deleted'}), 200
|