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

98.44

/freqtrade/configuration/load_config.py

"""
This module contain functions to load the configuration file
"""
import logging
import re
import sys
from copy import deepcopy
from pathlib import Path
from typing import Any, Dict, List, Optional

import rapidjson

from freqtrade.constants import MINIMAL_CONFIG, Config
from freqtrade.exceptions import ConfigurationError, OperationalException
from freqtrade.misc import deep_merge_dicts


logger = logging.getLogger(__name__)


CONFIG_PARSE_MODE = rapidjson.PM_COMMENTS | rapidjson.PM_TRAILING_COMMAS


def log_config_error_range(path: str, errmsg: str) -> str:
    """
    Parses configuration file and prints range around error
    """
    if path != '-':
        offsetlist = re.findall(r'(?<=Parse\serror\sat\soffset\s)\d+', errmsg)
        if offsetlist:
            offset = int(offsetlist[0])
            text = Path(path).read_text()
            # Fetch an offset of 80 characters around the error line
            subtext = text[offset - min(80, offset):offset + 80]
            segments = subtext.split('\n')
            if len(segments) > 3:
                # Remove first and last lines, to avoid odd truncations
                return '\n'.join(segments[1:-1])
            else:
                return subtext
    return ''


def load_file(path: Path) -> Dict[str, Any]:
    try:
        with path.open('r') as file:
            config = rapidjson.load(file, parse_mode=CONFIG_PARSE_MODE)
    except FileNotFoundError:
        raise OperationalException(f'File "{path}" not found!') from None
    return config


def load_config_file(path: str) -> Dict[str, Any]:
    """
    Loads a config file from the given path
    :param path: path as str
    :return: configuration as dictionary
    """
    try:
        # Read config from stdin if requested in the options
        with Path(path).open() if path != '-' else sys.stdin as file:
            config = rapidjson.load(file, parse_mode=CONFIG_PARSE_MODE)
    except FileNotFoundError:
        raise OperationalException(
            f'Config file "{path}" not found!'
            ' Please create a config file or check whether it exists.') from None
    except rapidjson.JSONDecodeError as e:
        err_range = log_config_error_range(path, str(e))
        raise ConfigurationError(
            f'{e}\n'
            f'Please verify the following segment of your configuration:\n{err_range}'
            if err_range else 'Please verify your configuration file for syntax errors.'
        )

    return config


def load_from_files(
        files: List[str], base_path: Optional[Path] = None, level: int = 0) -> Dict[str, Any]:
    """
    Recursively load configuration files if specified.
    Sub-files are assumed to be relative to the initial config.
    """
    config: Config = {}
    if level > 5:
        raise ConfigurationError("Config loop detected.")

    if not files:
        return deepcopy(MINIMAL_CONFIG)
    files_loaded = []
    # We expect here a list of config filenames
    for filename in files:
        logger.info(f'Using config: {filename} ...')
        if filename == '-':
            # Immediately load stdin and return
            return load_config_file(filename)
        file = Path(filename)
        if base_path:
            # Prepend basepath to allow for relative assignments
            file = base_path / file

        config_tmp = load_config_file(str(file))
        if 'add_config_files' in config_tmp:
            config_sub = load_from_files(
                config_tmp['add_config_files'], file.resolve().parent, level + 1)
            files_loaded.extend(config_sub.get('config_files', []))
            config_tmp = deep_merge_dicts(config_tmp, config_sub)

        files_loaded.insert(0, str(file))

        # Merge config options, overwriting prior values
        config = deep_merge_dicts(config_tmp, config)

    config['config_files'] = files_loaded

    return config

1	"""
2	This module contain functions to load the configuration file
3	"""
4	import logging	1✔
5	import re	1✔
6	import sys	1✔
7	from copy import deepcopy	1✔
8	from pathlib import Path	1✔
9	from typing import Any, Dict, List, Optional	1✔
10
11	import rapidjson	1✔
12
13	from freqtrade.constants import MINIMAL_CONFIG, Config	1✔
14	from freqtrade.exceptions import ConfigurationError, OperationalException	1✔
15	from freqtrade.misc import deep_merge_dicts	1✔
16
17
18	logger = logging.getLogger(__name__)	1✔
19
20
21	CONFIG_PARSE_MODE = rapidjson.PM_COMMENTS \| rapidjson.PM_TRAILING_COMMAS	1✔
22
23
24	def log_config_error_range(path: str, errmsg: str) -> str:	1✔
25	"""
26	Parses configuration file and prints range around error
27	"""
28	if path != '-':	1✔
29	offsetlist = re.findall(r'(?<=Parse\serror\sat\soffset\s)\d+', errmsg)	1✔
30	if offsetlist:	1✔
31	offset = int(offsetlist[0])	1✔
32	text = Path(path).read_text()	1✔
33	# Fetch an offset of 80 characters around the error line
34	subtext = text[offset - min(80, offset):offset + 80]	1✔
35	segments = subtext.split('\n')	1✔
36	if len(segments) > 3:	1✔
37	# Remove first and last lines, to avoid odd truncations
38	return '\n'.join(segments[1:-1])	1✔
39	else:
40	return subtext	1✔
41	return ''	1✔
42
43
44	def load_file(path: Path) -> Dict[str, Any]:	1✔
45	try:	1✔
46	with path.open('r') as file:	1✔
47	config = rapidjson.load(file, parse_mode=CONFIG_PARSE_MODE)	1✔
48	except FileNotFoundError:	1✔
49	raise OperationalException(f'File "{path}" not found!') from None	1✔
50	return config	1✔
51
52
53	def load_config_file(path: str) -> Dict[str, Any]:	1✔
54	"""
55	Loads a config file from the given path
56	:param path: path as str
57	:return: configuration as dictionary
58	"""
59	try:	1✔
60	# Read config from stdin if requested in the options
61	with Path(path).open() if path != '-' else sys.stdin as file:	1✔
62	config = rapidjson.load(file, parse_mode=CONFIG_PARSE_MODE)	1✔
63	except FileNotFoundError:	1✔
64	raise OperationalException(	1✔
65	f'Config file "{path}" not found!'
66	' Please create a config file or check whether it exists.') from None
67	except rapidjson.JSONDecodeError as e:	1✔
68	err_range = log_config_error_range(path, str(e))	1✔
69	raise ConfigurationError(	1✔
70	f'{e}\n'
71	f'Please verify the following segment of your configuration:\n{err_range}'
72	if err_range else 'Please verify your configuration file for syntax errors.'
73	)
74
75	return config	1✔
76
77
78	def load_from_files(	1✔
79	files: List[str], base_path: Optional[Path] = None, level: int = 0) -> Dict[str, Any]:
80	"""
81	Recursively load configuration files if specified.
82	Sub-files are assumed to be relative to the initial config.
83	"""
84	config: Config = {}	1✔
85	if level > 5:	1✔
86	raise ConfigurationError("Config loop detected.")	1✔
87
88	if not files:	1✔
89	return deepcopy(MINIMAL_CONFIG)	1✔
90	files_loaded = []	1✔
91	# We expect here a list of config filenames
92	for filename in files:	1✔
93	logger.info(f'Using config: {filename} ...')	1✔
94	if filename == '-':	1✔
95	# Immediately load stdin and return
96	return load_config_file(filename)	×
97	file = Path(filename)	1✔
98	if base_path:	1✔
99	# Prepend basepath to allow for relative assignments
100	file = base_path / file	1✔
101
102	config_tmp = load_config_file(str(file))	1✔
103	if 'add_config_files' in config_tmp:	1✔
104	config_sub = load_from_files(	1✔
105	config_tmp['add_config_files'], file.resolve().parent, level + 1)
106	files_loaded.extend(config_sub.get('config_files', []))	1✔
107	config_tmp = deep_merge_dicts(config_tmp, config_sub)	1✔
108
109	files_loaded.insert(0, str(file))	1✔
110
111	# Merge config options, overwriting prior values
112	config = deep_merge_dicts(config_tmp, config)	1✔
113
114	config['config_files'] = files_loaded	1✔
115
116	return config	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