83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
from typing import List
|
|
from pydantic import BaseModel
|
|
from app.services.database import get_db
|
|
import json
|
|
|
|
router = APIRouter()
|
|
|
|
class WebhookCreate(BaseModel):
|
|
project_id: int
|
|
url: str
|
|
events: List[str] = ["ticket.created", "ticket.updated", "comment.added"]
|
|
|
|
class WebhookResponse(BaseModel):
|
|
id: int
|
|
project_id: int
|
|
url: str
|
|
events: List[str]
|
|
active: bool
|
|
|
|
@router.get("", response_model=List[WebhookResponse])
|
|
async def list_webhooks(project_id: int = None):
|
|
db = await get_db()
|
|
if project_id:
|
|
cursor = await db.execute("SELECT * FROM webhooks WHERE project_id = ?", (project_id,))
|
|
else:
|
|
cursor = await db.execute("SELECT * FROM webhooks")
|
|
rows = await cursor.fetchall()
|
|
await db.close()
|
|
|
|
result = []
|
|
for row in rows:
|
|
wh = dict(row)
|
|
wh["events"] = json.loads(wh.get("events", "[]"))
|
|
wh["active"] = bool(wh.get("active", 1))
|
|
result.append(wh)
|
|
return result
|
|
|
|
@router.post("", response_model=WebhookResponse)
|
|
async def create_webhook(webhook: WebhookCreate):
|
|
db = await get_db()
|
|
cursor = await db.execute(
|
|
"INSERT INTO webhooks (project_id, url, events) VALUES (?, ?, ?)",
|
|
(webhook.project_id, webhook.url, json.dumps(webhook.events))
|
|
)
|
|
await db.commit()
|
|
webhook_id = cursor.lastrowid
|
|
|
|
cursor = await db.execute("SELECT * FROM webhooks WHERE id = ?", (webhook_id,))
|
|
row = await cursor.fetchone()
|
|
await db.close()
|
|
|
|
result = dict(row)
|
|
result["events"] = json.loads(result.get("events", "[]"))
|
|
result["active"] = bool(result.get("active", 1))
|
|
return result
|
|
|
|
@router.delete("/{webhook_id}")
|
|
async def delete_webhook(webhook_id: int):
|
|
db = await get_db()
|
|
await db.execute("DELETE FROM webhooks WHERE id = ?", (webhook_id,))
|
|
await db.commit()
|
|
await db.close()
|
|
return {"status": "deleted"}
|
|
|
|
@router.patch("/{webhook_id}/toggle")
|
|
async def toggle_webhook(webhook_id: int):
|
|
db = await get_db()
|
|
await db.execute("UPDATE webhooks SET active = NOT active WHERE id = ?", (webhook_id,))
|
|
await db.commit()
|
|
|
|
cursor = await db.execute("SELECT * FROM webhooks WHERE id = ?", (webhook_id,))
|
|
row = await cursor.fetchone()
|
|
await db.close()
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Webhook not found")
|
|
|
|
result = dict(row)
|
|
result["events"] = json.loads(result.get("events", "[]"))
|
|
result["active"] = bool(result.get("active", 1))
|
|
return result
|