70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
from sqlalchemy.orm import Session
|
|
from db.models.Path import Path
|
|
from db.models.Webhook import Webhook
|
|
from db.models.WebhookSetting import WebhookSetting
|
|
import abc
|
|
import importlib
|
|
import json
|
|
import pkgutil
|
|
import requests
|
|
import db
|
|
|
|
class WebhookEventHandler(abc.ABC):
|
|
def __init__(self, event_type=0):
|
|
self.event_type = event_type
|
|
|
|
@abc.abstractmethod
|
|
def get_path_id(self, payload):
|
|
pass
|
|
|
|
def __call__(self, payload):
|
|
path_id = self.get_path_id(payload)
|
|
with db.get_db() as session:
|
|
setting = self.get_setting(session, path_id)
|
|
if setting is None:
|
|
return
|
|
headers = {'Content-Type': 'application/json'}
|
|
if setting["additional_headers"] is not None:
|
|
headers.update(json.loads(setting["additional_headers"]))
|
|
try:
|
|
response = requests.post(setting["webhook_url"], json=payload, headers=headers, timeout=5)
|
|
response.raise_for_status()
|
|
except requests.RequestException as e:
|
|
print(e)
|
|
|
|
def get_setting(self, session: Session, path_id):
|
|
|
|
path = session.query(Path).filter(Path.id == path_id).first()
|
|
if path is None:
|
|
return None
|
|
p = path.to_dict()
|
|
webhook_setting = session.query(WebhookSetting).filter(WebhookSetting.path_id == path_id).first()
|
|
|
|
if webhook_setting is None and p["parent_id"] != 1:
|
|
return self.get_setting(session, p["parent_id"])
|
|
|
|
setting = webhook_setting.to_dict()
|
|
if not setting["enabled"] or setting["on_events"] & self.event_type == 0:
|
|
return None
|
|
webhook = session.query(Webhook).filter(Webhook.id == webhook_setting.webhook_id).first()
|
|
if webhook is None:
|
|
return None
|
|
setting["webhook_url"] = webhook.to_dict()["hook_url"]
|
|
return setting
|
|
|
|
_auto_instantiate_classes = set()
|
|
|
|
def auto_instantiate(cls):
|
|
_auto_instantiate_classes.add(cls)
|
|
return cls
|
|
|
|
def register_all_webhook_event_handlers():
|
|
package = __name__
|
|
package_path = __path__
|
|
|
|
for finder, name, ispkg in pkgutil.walk_packages(package_path, package+"."):
|
|
importlib.import_module(name)
|
|
|
|
for cls in _auto_instantiate_classes:
|
|
cls()
|