9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.06

/freqtrade/rpc/webhook.py

"""
This module manages webhook communication
"""
import logging
import time
from typing import Any, Dict, Optional

from requests import RequestException, post

from freqtrade.constants import Config
from freqtrade.enums import RPCMessageType
from freqtrade.rpc import RPC, RPCHandler
from freqtrade.rpc.rpc_types import RPCSendMsg


logger = logging.getLogger(__name__)

logger.debug('Included module rpc.webhook ...')


class Webhook(RPCHandler):
    """  This class handles all webhook communication """

    def __init__(self, rpc: RPC, config: Config) -> None:
        """
        Init the Webhook class, and init the super class RPCHandler
        :param rpc: instance of RPC Helper class
        :param config: Configuration object
        :return: None
        """
        super().__init__(rpc, config)

        self._url = self._config['webhook']['url']
        self._format = self._config['webhook'].get('format', 'form')
        self._retries = self._config['webhook'].get('retries', 0)
        self._retry_delay = self._config['webhook'].get('retry_delay', 0.1)
        self._timeout = self._config['webhook'].get('timeout', 10)

    def cleanup(self) -> None:
        """
        Cleanup pending module resources.
        This will do nothing for webhooks, they will simply not be called anymore
        """
        pass

    def _get_value_dict(self, msg: RPCSendMsg) -> Optional[Dict[str, Any]]:
        whconfig = self._config['webhook']
        if msg['type'].value in whconfig:
            # Explicit types should have priority
            valuedict = whconfig.get(msg['type'].value)
        # Deprecated 2022.10 - only keep generic method.
        elif msg['type'] in [RPCMessageType.ENTRY]:
            valuedict = whconfig.get('webhookentry')
        elif msg['type'] in [RPCMessageType.ENTRY_CANCEL]:
            valuedict = whconfig.get('webhookentrycancel')
        elif msg['type'] in [RPCMessageType.ENTRY_FILL]:
            valuedict = whconfig.get('webhookentryfill')
        elif msg['type'] == RPCMessageType.EXIT:
            valuedict = whconfig.get('webhookexit')
        elif msg['type'] == RPCMessageType.EXIT_FILL:
            valuedict = whconfig.get('webhookexitfill')
        elif msg['type'] == RPCMessageType.EXIT_CANCEL:
            valuedict = whconfig.get('webhookexitcancel')
        elif msg['type'] in (RPCMessageType.STATUS,
                             RPCMessageType.STARTUP,
                             RPCMessageType.EXCEPTION,
                             RPCMessageType.WARNING):
            valuedict = whconfig.get('webhookstatus')
        elif msg['type'] in (
                RPCMessageType.PROTECTION_TRIGGER,
                RPCMessageType.PROTECTION_TRIGGER_GLOBAL,
                RPCMessageType.WHITELIST,
                RPCMessageType.ANALYZED_DF,
                RPCMessageType.NEW_CANDLE,
                RPCMessageType.STRATEGY_MSG):
            # Don't fail for non-implemented types
            return None
        return valuedict

    def send_msg(self, msg: RPCSendMsg) -> None:
        """ Send a message to telegram channel """
        try:

            valuedict = self._get_value_dict(msg)

            if not valuedict:
                logger.debug("Message type '%s' not configured for webhooks", msg['type'])
                return

            payload = {key: value.format(**msg) for (key, value) in valuedict.items()}
            self._send_msg(payload)
        except KeyError as exc:
            logger.exception("Problem calling Webhook. Please check your webhook configuration. "
                             "Exception: %s", exc)

    def _send_msg(self, payload: dict) -> None:
        """do the actual call to the webhook"""

        success = False
        attempts = 0
        while not success and attempts <= self._retries:
            if attempts:
                if self._retry_delay:
                    time.sleep(self._retry_delay)
                logger.info("Retrying webhook...")

            attempts += 1

            try:
                if self._format == 'form':
                    response = post(self._url, data=payload, timeout=self._timeout)
                elif self._format == 'json':
                    response = post(self._url, json=payload, timeout=self._timeout)
                elif self._format == 'raw':
                    response = post(self._url, data=payload['data'],
                                    headers={'Content-Type': 'text/plain'},
                                    timeout=self._timeout)
                else:
                    raise NotImplementedError(f'Unknown format: {self._format}')

                # Throw a RequestException if the post was not successful
                response.raise_for_status()
                success = True

            except RequestException as exc:
                logger.warning("Could not call webhook url. Exception: %s", exc)

1	"""
2	This module manages webhook communication
3	"""
4	import logging	1✔
5	import time	1✔
6	from typing import Any, Dict, Optional	1✔
7
8	from requests import RequestException, post	1✔
9
10	from freqtrade.constants import Config	1✔
11	from freqtrade.enums import RPCMessageType	1✔
12	from freqtrade.rpc import RPC, RPCHandler	1✔
13	from freqtrade.rpc.rpc_types import RPCSendMsg	1✔
14
15
16	logger = logging.getLogger(__name__)	1✔
17
18	logger.debug('Included module rpc.webhook ...')	1✔
19
20
21	class Webhook(RPCHandler):	1✔
22	""" This class handles all webhook communication """
23
24	def __init__(self, rpc: RPC, config: Config) -> None:	1✔
25	"""
26	Init the Webhook class, and init the super class RPCHandler
27	:param rpc: instance of RPC Helper class
28	:param config: Configuration object
29	:return: None
30	"""
31	super().__init__(rpc, config)	1✔
32
33	self._url = self._config['webhook']['url']	1✔
34	self._format = self._config['webhook'].get('format', 'form')	1✔
35	self._retries = self._config['webhook'].get('retries', 0)	1✔
36	self._retry_delay = self._config['webhook'].get('retry_delay', 0.1)	1✔
37	self._timeout = self._config['webhook'].get('timeout', 10)	1✔
38
39	def cleanup(self) -> None:	1✔
40	"""
41	Cleanup pending module resources.
42	This will do nothing for webhooks, they will simply not be called anymore
43	"""
44	pass	×
45
46	def _get_value_dict(self, msg: RPCSendMsg) -> Optional[Dict[str, Any]]:	1✔
47	whconfig = self._config['webhook']	1✔
48	if msg['type'].value in whconfig:	1✔
49	# Explicit types should have priority
50	valuedict = whconfig.get(msg['type'].value)	1✔
51	# Deprecated 2022.10 - only keep generic method.
52	elif msg['type'] in [RPCMessageType.ENTRY]:	1✔
53	valuedict = whconfig.get('webhookentry')	1✔
54	elif msg['type'] in [RPCMessageType.ENTRY_CANCEL]:	1✔
55	valuedict = whconfig.get('webhookentrycancel')	1✔
56	elif msg['type'] in [RPCMessageType.ENTRY_FILL]:	1✔
57	valuedict = whconfig.get('webhookentryfill')	1✔
58	elif msg['type'] == RPCMessageType.EXIT:	1✔
59	valuedict = whconfig.get('webhookexit')	1✔
60	elif msg['type'] == RPCMessageType.EXIT_FILL:	1✔
61	valuedict = whconfig.get('webhookexitfill')	1✔
62	elif msg['type'] == RPCMessageType.EXIT_CANCEL:	1✔
63	valuedict = whconfig.get('webhookexitcancel')	1✔
64	elif msg['type'] in (RPCMessageType.STATUS,	1✔
65	RPCMessageType.STARTUP,
66	RPCMessageType.EXCEPTION,
67	RPCMessageType.WARNING):
68	valuedict = whconfig.get('webhookstatus')	1✔
69	elif msg['type'] in (	1✔
70	RPCMessageType.PROTECTION_TRIGGER,
71	RPCMessageType.PROTECTION_TRIGGER_GLOBAL,
72	RPCMessageType.WHITELIST,
73	RPCMessageType.ANALYZED_DF,
74	RPCMessageType.NEW_CANDLE,
75	RPCMessageType.STRATEGY_MSG):
76	# Don't fail for non-implemented types
77	return None	1✔
78	return valuedict	1✔
79
80	def send_msg(self, msg: RPCSendMsg) -> None:	1✔
81	""" Send a message to telegram channel """
82	try:	1✔
83
84	valuedict = self._get_value_dict(msg)	1✔
85
86	if not valuedict:	1✔
87	logger.debug("Message type '%s' not configured for webhooks", msg['type'])	1✔
88	return	1✔
89
90	payload = {key: value.format(**msg) for (key, value) in valuedict.items()}	1✔
91	self._send_msg(payload)	1✔
92	except KeyError as exc:	1✔
93	logger.exception("Problem calling Webhook. Please check your webhook configuration. "	1✔
94	"Exception: %s", exc)
95
96	def _send_msg(self, payload: dict) -> None:	1✔
97	"""do the actual call to the webhook"""
98
99	success = False	1✔
100	attempts = 0	1✔
101	while not success and attempts <= self._retries:	1✔
102	if attempts:	1✔
103	if self._retry_delay:	×
104	time.sleep(self._retry_delay)	×
105	logger.info("Retrying webhook...")	×
106
107	attempts += 1	1✔
108
109	try:	1✔
110	if self._format == 'form':	1✔
111	response = post(self._url, data=payload, timeout=self._timeout)	1✔
112	elif self._format == 'json':	1✔
113	response = post(self._url, json=payload, timeout=self._timeout)	1✔
114	elif self._format == 'raw':	1✔
115	response = post(self._url, data=payload['data'],	1✔
116	headers={'Content-Type': 'text/plain'},
117	timeout=self._timeout)
118	else:
119	raise NotImplementedError(f'Unknown format: {self._format}')	×
120
121	# Throw a RequestException if the post was not successful
122	response.raise_for_status()	1✔
123	success = True	1✔
124
125	except RequestException as exc:	1✔
126	logger.warning("Could not call webhook url. Exception: %s", exc)	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous