1import json 

2import yaml 

3import requests 

4 

5webhook_schema = yaml.safe_load(""" 

6type: object 

7properties: 

8 url: {type: string} 

9 query: {type: object} 

10 body: {type: object} 

11 method: {type: string} 

12 cookies: {type: object} 

13 headers: {type: object} 

14required: [url, body] 

15""") 

16 

17 

18def Base_webhook(self, params): 

19 """ 

20 - name: notify webhook 

21 webhook: 

22 url: https://host/path 

23 query: 

24 id: "xyz" 

25 body: 

26 text: "hello, world" 

27 """ 

28 url = params.get("url") 

29 query = params.get("query", {}) 

30 body = params.get("body", {}) 

31 method = params.get("method", "post") 

32 cookies = params.get("cookies", {}) 

33 headers = params.get("headers", {"content-type": "application/json"}) 

34 timeout = params.get("timeout", None) 

35 sess = requests.Session() 

36 for name, value in cookies.items(): 

37 sess.cookies.set(name, value) 

38 resp = sess.request(method, url, params=query, headers=headers, timeout=timeout, 

39 data=json.dumps(body, ensure_ascii=False)) 

40 return resp.json()