655228070

Build Type

github

Committed by GitHub

Commit Message

Merge a8d7e6584 into 8dcb452c5

Pull Request Pull Request #312: feat: cache remote styles to avoid HTTP requests

Run Details

451 of 511 branches covered (88.26%)

Branch coverage included in aggregate %.

60 of 62 new or added lines in 5 files covered. (96.77%)

27 existing lines in 3 files now uncovered.

1593 of 1655 relevant lines covered (96.25%)

3.84 hits per line

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

86.97

/src/nitpick/style.py

"""Style files."""
from collections import OrderedDict
from datetime import datetime, timedelta
from functools import lru_cache
from pathlib import Path
from typing import Dict, Iterator, List, Optional, Set, Tuple, Type
from urllib.parse import urlparse, urlunparse

import click
import requests
from cachy import CacheManager, Repository
from identify import identify
from loguru import logger
from marshmallow import Schema
from slugify import slugify
from toml import TomlDecodeError

from nitpick import __version__, fields
from nitpick.constants import (
    CACHE_DIR_NAME,
    DOT_SLASH,
    MERGED_STYLE_TOML,
    NITPICK_STYLE_TOML,
    NITPICK_STYLES_INCLUDE_JMEX,
    PROJECT_NAME,
    PYPROJECT_TOML,
    RAW_GITHUB_CONTENT_BASE_URL,
    TOML_EXTENSION,
)
from nitpick.enums import CachingEnum, OptionEnum
from nitpick.exceptions import Deprecation, QuitComplainingError, pretty_exception
from nitpick.formats import TOMLFormat
from nitpick.generic import MergeDict, is_url, search_dict
from nitpick.plugins.base import NitpickPlugin
from nitpick.plugins.info import FileInfo
from nitpick.project import Project, climb_directory_tree
from nitpick.schemas import BaseStyleSchema, NitpickSectionSchema, flatten_marshmallow_errors
from nitpick.typedefs import JsonDict, StrOrList, mypy_property
from nitpick.violations import Fuss, Reporter, StyleViolations

Plugins = Set[Type[NitpickPlugin]]


class Style:  # pylint: disable=too-many-instance-attributes
    """Include styles recursively from one another."""

    def __init__(self, project: Project, offline: bool, caching: CachingEnum, caching_delta: timedelta = None) -> None:
        self.project: Project = project
        self.offline = offline
        self.caching = caching
        self.caching_delta: timedelta = caching_delta or timedelta()

        self._all_styles = MergeDict()
        self._already_included: Set[str] = set()
        self._first_full_path: str = ""

        self._dynamic_schema_class: type = BaseStyleSchema
        self.rebuild_dynamic_schema()

    @mypy_property
    @lru_cache()
    def cache_dir(self) -> Path:
        """Clear the cache directory (on the project root or on the current directory)."""
        path: Path = self.project.root / CACHE_DIR_NAME / PROJECT_NAME

        # TODO: check if the merged style file is still needed; if not, this line can be removed
        path.mkdir(parents=True, exist_ok=True)

        return path

    @property
    def cache_manager(self) -> Repository:
        """Return a cache manager for the cache directory.

        `File store <https://cachy.readthedocs.io/en/latest/configuration.html#file>`_.
        """
        return CacheManager({"stores": {"file": {"driver": "file", "path": self.cache_dir}}}).store()

    @staticmethod
    def get_default_style_url():
        """Return the URL of the default style for the current version."""
        return f"{RAW_GITHUB_CONTENT_BASE_URL}v{__version__}/{NITPICK_STYLE_TOML}"

    def find_initial_styles(self, configured_styles: StrOrList) -> Iterator[Fuss]:
        """Find the initial style(s) and include them."""
        if configured_styles:
            chosen_styles = configured_styles
            log_message = f"Styles configured in {PYPROJECT_TOML}"
        else:
            paths = climb_directory_tree(self.project.root, [NITPICK_STYLE_TOML])
            if paths:
                chosen_styles = str(sorted(paths)[0])
                log_message = "Found style climbing the directory tree"
            else:
                chosen_styles = self.get_default_style_url()
                log_message = "Loading default Nitpick style"
        logger.info(f"{log_message}: {chosen_styles}")

        yield from self.include_multiple_styles(chosen_styles)

    @staticmethod
    def validate_schema(schema: Type[Schema], path_from_root: str, original_data: JsonDict) -> Dict[str, List[str]]:
        """Validate the schema for the file."""
        if not schema:
            return {}

        inherited_schema = schema is not BaseStyleSchema
        data_to_validate = original_data if inherited_schema else {path_from_root: None}
        local_errors = schema().validate(data_to_validate)
        if local_errors and inherited_schema:
            local_errors = {path_from_root: local_errors}
        return local_errors

    def include_multiple_styles(self, chosen_styles: StrOrList) -> Iterator[Fuss]:
        """Include a list of styles (or just one) into this style tree."""
        style_uris: List[str] = [chosen_styles] if isinstance(chosen_styles, str) else chosen_styles
        for style_uri in style_uris:
            style_path, file_contents = self.get_style_path(style_uri)
            if not style_path:
                continue

            toml = TOMLFormat(string=file_contents)
            try:
                read_toml_dict = toml.as_data
            except TomlDecodeError as err:
                # If the TOML itself could not be parsed, we can't go on
                raise QuitComplainingError(
                    Reporter(FileInfo(self.project, style_path.name)).make_fuss(
                        StyleViolations.INVALID_TOML, exception=pretty_exception(err)
                    )
                ) from err

            try:
                display_name = str(style_path.relative_to(self.project.root))
            except ValueError:
                display_name = style_uri

            toml_dict, validation_errors = self._validate_config(read_toml_dict)

            if validation_errors:
                yield Reporter(FileInfo(self.project, display_name)).make_fuss(
                    StyleViolations.INVALID_CONFIG, flatten_marshmallow_errors(validation_errors)
                )

            self._all_styles.add(toml_dict)

            sub_styles: StrOrList = search_dict(NITPICK_STYLES_INCLUDE_JMEX, toml_dict, [])
            if sub_styles:
                yield from self.include_multiple_styles(sub_styles)

    def _validate_config(self, config_dict: Dict) -> Tuple[Dict, Dict]:
        validation_errors = {}
        toml_dict = OrderedDict()
        for key, value_dict in config_dict.items():
            info = FileInfo.create(self.project, key)
            toml_dict[info.path_from_root] = value_dict
            if key == PROJECT_NAME:
                schemas = [NitpickSectionSchema]
            else:
                schemas = [
                    plugin_class.validation_schema
                    for plugin_class in self.project.plugin_manager.hook.can_handle(
                        info=info
                    )  # pylint: disable=no-member
                ]
                if not schemas:
                    validation_errors[key] = [BaseStyleSchema.error_messages["unknown"]]

            all_errors = {}
            valid_schema = False
            for schema in schemas:
                errors = self.validate_schema(schema, info.path_from_root, value_dict)
                if not errors:
                    # When multiple schemas match a file type, exit when a valid schema is found
                    valid_schema = True
                    break
                all_errors.update(errors)

            if not valid_schema:
                Deprecation.jsonfile_section(all_errors)
                validation_errors.update(all_errors)
        return toml_dict, validation_errors

    def get_style_path(self, style_uri: str) -> Tuple[Optional[Path], str]:
        """Get the style path from the URI. Add the .toml extension if it's missing."""
        clean_style_uri = style_uri.strip()

        remote = False
        if clean_style_uri.startswith(DOT_SLASH):
            remote = False
        elif is_url(clean_style_uri) or is_url(self._first_full_path):
            remote = True

        if remote is True:
            return self.fetch_style_from_url(clean_style_uri)
        return self.fetch_style_from_local_path(clean_style_uri)

    def fetch_style_from_url(self, url: str) -> Tuple[Optional[Path], str]:
        """Fetch a style file from a URL, saving the contents in the cache dir."""
        if self.offline:
            # No style will be fetched in offline mode
            return None, ""

        if self._first_full_path and not is_url(url):
            prefix, rest = self._first_full_path.split(":/")
            domain_plus_url = str(rest).strip("/").rstrip("/") + "/" + url
            new_url = f"{prefix}://{domain_plus_url}"
        else:
            new_url = url

        parsed_url = list(urlparse(new_url))
        if not parsed_url[2].endswith(TOML_EXTENSION):
            parsed_url[2] += TOML_EXTENSION
        new_url = urlunparse(parsed_url)

        if new_url in self._already_included:
            return None, ""

        try:
            contents = self.fetch_from_url_or_cache(new_url)
        except requests.ConnectionError:
            click.secho(
                "Your network is unreachable. Fix your connection or use"
                f" {OptionEnum.OFFLINE.as_flake8_flag()} / {OptionEnum.OFFLINE.as_envvar()}=1",
                fg="red",
                err=True,
            )
            return None, ""

        # Save the first full path to be used by the next files without parent.
        if not self._first_full_path:
            self._first_full_path = new_url.rsplit("/", 1)[0]

        self._already_included.add(new_url)
        return Path(f"{slugify(new_url)}.toml"), contents

    def fetch_from_url_or_cache(self, new_url: str) -> str:
        """Fetch data from a remote URL, or read from the local cache.

        `Storing Items In The Cache <https://cachy.readthedocs.io/en/latest/usage.html#storing-items-in-the-cache>`_.
        """
        if self.caching != CachingEnum.NEVER:
            cached_value = self.cache_manager.get(new_url)
            if cached_value is not None:
                logger.debug(f"Using cached value for URL {new_url}")
                return cached_value

        response = requests.get(new_url)
        logger.debug(f"Requesting style from URL {new_url}")
        if not response.ok:
            raise FileNotFoundError(f"Error {response} fetching style URL {new_url}")

        contents = response.text

        if self.caching == CachingEnum.FOREVER:
            self.cache_manager.forever(new_url, contents)
        elif self.caching == CachingEnum.EXPIRES:
            self.cache_manager.put(new_url, contents, datetime.now() + self.caching_delta)

        return contents

    def fetch_style_from_local_path(self, partial_filename: str) -> Tuple[Optional[Path], str]:
        """Fetch a style file from a local path."""
        if partial_filename and not partial_filename.endswith(TOML_EXTENSION):
            partial_filename += TOML_EXTENSION
        expanded_path = Path(partial_filename).expanduser()

        if self._first_full_path and not (str(expanded_path).startswith("/") or partial_filename.startswith(DOT_SLASH)):
            # Prepend the previous path to the partial file name.
            style_path = Path(self._first_full_path) / expanded_path
        else:
            # Get the absolute path, be it from a root path (starting with slash) or from the current dir.
            style_path = Path(expanded_path).absolute()

        # Save the first full path to be used by the next files without parent.
        if not self._first_full_path:
            self._first_full_path = str(style_path.resolve().parent)

        if str(style_path) in self._already_included:
            return None, ""

        if not style_path.exists():
            raise FileNotFoundError(f"Local style file does not exist: {style_path}")

        logger.debug(f"Loading style from {style_path}")
        self._already_included.add(str(style_path))
        return style_path, style_path.read_text()

    def merge_toml_dict(self) -> JsonDict:
        """Merge all included styles into a TOML (actually JSON) dictionary."""
        merged_dict = self._all_styles.merge()
        # TODO: check if the merged style file is still needed
        merged_style_path: Path = self.cache_dir / MERGED_STYLE_TOML
        toml = TOMLFormat(data=merged_dict)

        attempt = 1
        while attempt < 5:
            try:
                merged_style_path.write_text(toml.reformatted)
                break
            except OSError:
                attempt += 1

        return merged_dict

    @staticmethod
    def file_field_pair(filename: str, base_file_class: Type[NitpickPlugin]) -> Dict[str, fields.Field]:
        """Return a schema field with info from a config file class."""
        unique_filename_with_underscore = slugify(filename, separator="_")

        kwargs = {"data_key": filename}
        if base_file_class.validation_schema:
            field = fields.Nested(base_file_class.validation_schema, **kwargs)
        else:
            # For some files (e.g.: pyproject.toml, INI files), there is no strict schema;
            # it can be anything they allow.
            # It's out of Nitpick's scope to validate those files.
            field = fields.Dict(fields.String, **kwargs)
        return {unique_filename_with_underscore: field}

    @lru_cache()
    def load_fixed_dynamic_classes(self) -> Tuple[Plugins, Plugins]:
        """Separate classes with fixed file names from classes with dynamic files names."""
        fixed_name_classes: Plugins = set()
        dynamic_name_classes: Plugins = set()
        for plugin_class in self.project.plugin_manager.hook.plugin_class():  # pylint: disable=no-member
            if plugin_class.filename:
                fixed_name_classes.add(plugin_class)
            else:
                dynamic_name_classes.add(plugin_class)
        return fixed_name_classes, dynamic_name_classes

    def rebuild_dynamic_schema(self, data: JsonDict = None) -> None:
        """Rebuild the dynamic Marshmallow schema when needed, adding new fields that were found on the style."""
        new_files_found: Dict[str, fields.Field] = OrderedDict()

        fixed_name_classes, dynamic_name_classes = self.load_fixed_dynamic_classes()

        if data is None:
            # Data is empty; so this is the first time the dynamic class is being rebuilt.
            # Loop on classes with predetermined names, and add fields for them on the dynamic validation schema.
            # E.g.: pre-commit, pyproject.toml: files whose names we already know at this point.
            for subclass in fixed_name_classes:
                new_files_found.update(self.file_field_pair(subclass.filename, subclass))
        else:
            handled_tags: Dict[str, Type[NitpickPlugin]] = {}

            # Data was provided; search it to find new dynamic files to add to the validation schema).
            # E.g.: JSON files that were configured on some TOML style file.
            for subclass in dynamic_name_classes:
                for tag in subclass.identify_tags:
                    # TODO: WRONG! a tag should be handled by multiple classes...
                    # A tag can only be handled by a single subclass.
                    # If more than one class handle a tag, the latest one will be the handler.
                    handled_tags[tag] = subclass

                jmex = subclass.get_compiled_jmespath_filenames()
                for configured_filename in search_dict(jmex, data, []):
                    new_files_found.update(self.file_field_pair(configured_filename, subclass))

            self._find_subclasses(data, handled_tags, new_files_found)

        # Only recreate the schema if new fields were found.
        if new_files_found:
            self._dynamic_schema_class = type("DynamicStyleSchema", (self._dynamic_schema_class,), new_files_found)

    def _find_subclasses(self, data, handled_tags, new_files_found):
        for possible_file in data.keys():
            found_subclasses = []
            for file_tag in identify.tags_from_filename(possible_file):
                handler_subclass = handled_tags.get(file_tag)
                if handler_subclass:
                    found_subclasses.append(handler_subclass)

            for found_subclass in found_subclasses:
                new_files_found.update(self.file_field_pair(possible_file, found_subclass))

1	"""Style files."""
2	from collections import OrderedDict	4✔
3	from datetime import datetime, timedelta	4✔
4	from functools import lru_cache	4✔
5	from pathlib import Path	4✔
6	from typing import Dict, Iterator, List, Optional, Set, Tuple, Type	4✔
7	from urllib.parse import urlparse, urlunparse	4✔
8
9	import click	4✔
10	import requests	4✔
11	from cachy import CacheManager, Repository	4✔
12	from identify import identify	4✔
13	from loguru import logger	4✔
14	from marshmallow import Schema	4✔
15	from slugify import slugify	4✔
16	from toml import TomlDecodeError	4✔
17
18	from nitpick import __version__, fields	4✔
19	from nitpick.constants import (	4✔
20	CACHE_DIR_NAME,
21	DOT_SLASH,
22	MERGED_STYLE_TOML,
23	NITPICK_STYLE_TOML,
24	NITPICK_STYLES_INCLUDE_JMEX,
25	PROJECT_NAME,
26	PYPROJECT_TOML,
27	RAW_GITHUB_CONTENT_BASE_URL,
28	TOML_EXTENSION,
29	)
30	from nitpick.enums import CachingEnum, OptionEnum	4✔
31	from nitpick.exceptions import Deprecation, QuitComplainingError, pretty_exception	4✔
32	from nitpick.formats import TOMLFormat	4✔
33	from nitpick.generic import MergeDict, is_url, search_dict	4✔
34	from nitpick.plugins.base import NitpickPlugin	4✔
35	from nitpick.plugins.info import FileInfo	4✔
36	from nitpick.project import Project, climb_directory_tree	4✔
37	from nitpick.schemas import BaseStyleSchema, NitpickSectionSchema, flatten_marshmallow_errors	4✔
38	from nitpick.typedefs import JsonDict, StrOrList, mypy_property	4✔
39	from nitpick.violations import Fuss, Reporter, StyleViolations	4✔
40
41	Plugins = Set[Type[NitpickPlugin]]	4✔
42
43
44	class Style: # pylint: disable=too-many-instance-attributes	4✔
45	"""Include styles recursively from one another."""
46
47	def __init__(self, project: Project, offline: bool, caching: CachingEnum, caching_delta: timedelta = None) -> None:	4✔
48	self.project: Project = project	4✔
49	self.offline = offline	4✔
50	self.caching = caching	4✔
51	self.caching_delta: timedelta = caching_delta or timedelta()	4✔
52
53	self._all_styles = MergeDict()	4✔
54	self._already_included: Set[str] = set()	4✔
55	self._first_full_path: str = ""	4✔
56
57	self._dynamic_schema_class: type = BaseStyleSchema	4✔
58	self.rebuild_dynamic_schema()	4✔
59
60	@mypy_property	4✔
61	@lru_cache()	4✔
62	def cache_dir(self) -> Path:	4✔
63	"""Clear the cache directory (on the project root or on the current directory)."""
64	path: Path = self.project.root / CACHE_DIR_NAME / PROJECT_NAME	4✔
65
66	# TODO: check if the merged style file is still needed; if not, this line can be removed
67	path.mkdir(parents=True, exist_ok=True)	4✔
68
69	return path	4✔
70
71	@property	4✔
72	def cache_manager(self) -> Repository:	4✔
73	"""Return a cache manager for the cache directory.
74
75	`File store <https://cachy.readthedocs.io/en/latest/configuration.html#file>`_.
76	"""
77	return CacheManager({"stores": {"file": {"driver": "file", "path": self.cache_dir}}}).store()	4✔
78
79	@staticmethod	4✔
80	def get_default_style_url():	2✔
81	"""Return the URL of the default style for the current version."""
82	return f"{RAW_GITHUB_CONTENT_BASE_URL}v{__version__}/{NITPICK_STYLE_TOML}"	4✔
83
84	def find_initial_styles(self, configured_styles: StrOrList) -> Iterator[Fuss]:	4✔
85	"""Find the initial style(s) and include them."""
86	if configured_styles:	4✔
87	chosen_styles = configured_styles	4✔
88	log_message = f"Styles configured in {PYPROJECT_TOML}"	4✔
89	else:
90	paths = climb_directory_tree(self.project.root, [NITPICK_STYLE_TOML])	4✔
91	if paths:	4✔
92	chosen_styles = str(sorted(paths)[0])	4✔
93	log_message = "Found style climbing the directory tree"	4✔
94	else:
95	chosen_styles = self.get_default_style_url()	4✔
96	log_message = "Loading default Nitpick style"	4✔
97	logger.info(f"{log_message}: {chosen_styles}")	4✔
98
99	yield from self.include_multiple_styles(chosen_styles)	4✔
100
101	@staticmethod	4✔
102	def validate_schema(schema: Type[Schema], path_from_root: str, original_data: JsonDict) -> Dict[str, List[str]]:	4✔
103	"""Validate the schema for the file."""
104	if not schema:	4✔
105	return {}	4✔
106
107	inherited_schema = schema is not BaseStyleSchema	4✔
108	data_to_validate = original_data if inherited_schema else {path_from_root: None}	4✔
109	local_errors = schema().validate(data_to_validate)	4✔
110	if local_errors and inherited_schema:	4✔
111	local_errors = {path_from_root: local_errors}	4✔
112	return local_errors	4✔
113
114	def include_multiple_styles(self, chosen_styles: StrOrList) -> Iterator[Fuss]:	4✔
115	"""Include a list of styles (or just one) into this style tree."""
116	style_uris: List[str] = [chosen_styles] if isinstance(chosen_styles, str) else chosen_styles	4✔
117	for style_uri in style_uris:	4✔
118	style_path, file_contents = self.get_style_path(style_uri)	4✔
119	if not style_path:	4✔
120	continue	4✔
121
122	toml = TOMLFormat(string=file_contents)	4✔
123	try:	4✔
124	read_toml_dict = toml.as_data	4✔
125	except TomlDecodeError as err:	4✔
126	# If the TOML itself could not be parsed, we can't go on
127	raise QuitComplainingError(	4✔
128	Reporter(FileInfo(self.project, style_path.name)).make_fuss(
129	StyleViolations.INVALID_TOML, exception=pretty_exception(err)
130	)
131	) from err
132
133	try:	4✔
134	display_name = str(style_path.relative_to(self.project.root))	4✔
135	except ValueError:	4✔
136	display_name = style_uri	4✔
137
138	toml_dict, validation_errors = self._validate_config(read_toml_dict)	4✔
139
140	if validation_errors:	4✔
141	yield Reporter(FileInfo(self.project, display_name)).make_fuss(	4✔
142	StyleViolations.INVALID_CONFIG, flatten_marshmallow_errors(validation_errors)
143	)
144
145	self._all_styles.add(toml_dict)	4✔
146
147	sub_styles: StrOrList = search_dict(NITPICK_STYLES_INCLUDE_JMEX, toml_dict, [])	4✔
148	if sub_styles:	4✔
149	yield from self.include_multiple_styles(sub_styles)	4✔
150
151	def _validate_config(self, config_dict: Dict) -> Tuple[Dict, Dict]:	4✔
152	validation_errors = {}	4✔
153	toml_dict = OrderedDict()	4✔
154	for key, value_dict in config_dict.items():	4✔
155	info = FileInfo.create(self.project, key)	4✔
156	toml_dict[info.path_from_root] = value_dict	4✔
157	if key == PROJECT_NAME:	4✔
158	schemas = [NitpickSectionSchema]	4✔
159	else:
160	schemas = [	4✔
161	plugin_class.validation_schema
162	for plugin_class in self.project.plugin_manager.hook.can_handle(
163	info=info
164	) # pylint: disable=no-member
165	]
166	if not schemas:	4✔
167	validation_errors[key] = [BaseStyleSchema.error_messages["unknown"]]	4✔
168
169	all_errors = {}	4✔
170	valid_schema = False	4✔
171	for schema in schemas:	4✔
172	errors = self.validate_schema(schema, info.path_from_root, value_dict)	4✔
173	if not errors:	4✔
174	# When multiple schemas match a file type, exit when a valid schema is found
175	valid_schema = True	4✔
176	break	4✔
177	all_errors.update(errors)	4✔
178
179	if not valid_schema:	4✔
180	Deprecation.jsonfile_section(all_errors)	4✔
181	validation_errors.update(all_errors)	4✔
182	return toml_dict, validation_errors	4✔
183
184	def get_style_path(self, style_uri: str) -> Tuple[Optional[Path], str]:	4✔
185	"""Get the style path from the URI. Add the .toml extension if it's missing."""
186	clean_style_uri = style_uri.strip()	4✔
187
188	remote = False	4✔
189	if clean_style_uri.startswith(DOT_SLASH):	4✔
190	remote = False	4✔
191	elif is_url(clean_style_uri) or is_url(self._first_full_path):	4✔
192	remote = True	4✔
193
194	if remote is True:	4✔
195	return self.fetch_style_from_url(clean_style_uri)	4✔
196	return self.fetch_style_from_local_path(clean_style_uri)	4✔
197
198	def fetch_style_from_url(self, url: str) -> Tuple[Optional[Path], str]:	4✔
199	"""Fetch a style file from a URL, saving the contents in the cache dir."""
200	if self.offline:	4✔
201	# No style will be fetched in offline mode
202	return None, ""	4✔
203
204	if self._first_full_path and not is_url(url):	4✔
205	prefix, rest = self._first_full_path.split(":/")	4✔
206	domain_plus_url = str(rest).strip("/").rstrip("/") + "/" + url	4✔
207	new_url = f"{prefix}://{domain_plus_url}"	4✔
208	else:
209	new_url = url	4✔
210
211	parsed_url = list(urlparse(new_url))	4✔
212	if not parsed_url[2].endswith(TOML_EXTENSION):	4✔
213	parsed_url[2] += TOML_EXTENSION	4✔
214	new_url = urlunparse(parsed_url)	4✔
215
216	if new_url in self._already_included:	4!
NEW UNCOV 217	return None, ""	×
218
219	try:	4✔
220	contents = self.fetch_from_url_or_cache(new_url)	4✔
221	except requests.ConnectionError:	4✔
222	click.secho(	4✔
223	"Your network is unreachable. Fix your connection or use"
224	f" {OptionEnum.OFFLINE.as_flake8_flag()} / {OptionEnum.OFFLINE.as_envvar()}=1",
225	fg="red",
226	err=True,
227	)
228	return None, ""	4✔
229
230	# Save the first full path to be used by the next files without parent.
231	if not self._first_full_path:	4✔
232	self._first_full_path = new_url.rsplit("/", 1)[0]	4✔
233
234	self._already_included.add(new_url)	4✔
235	return Path(f"{slugify(new_url)}.toml"), contents	4✔
236
237	def fetch_from_url_or_cache(self, new_url: str) -> str:	4✔
238	"""Fetch data from a remote URL, or read from the local cache.
239
240	`Storing Items In The Cache <https://cachy.readthedocs.io/en/latest/usage.html#storing-items-in-the-cache>`_.
241	"""
242	if self.caching != CachingEnum.NEVER:	4✔
243	cached_value = self.cache_manager.get(new_url)	4✔
244	if cached_value is not None:	4✔
245	logger.debug(f"Using cached value for URL {new_url}")	4✔
246	return cached_value	4✔
247
248	response = requests.get(new_url)	4✔
249	logger.debug(f"Requesting style from URL {new_url}")	4✔
250	if not response.ok:	4!
NEW UNCOV 251	raise FileNotFoundError(f"Error {response} fetching style URL {new_url}")	×
252
253	contents = response.text	4✔
254
255	if self.caching == CachingEnum.FOREVER:	4✔
256	self.cache_manager.forever(new_url, contents)	4✔
257	elif self.caching == CachingEnum.EXPIRES:	4✔
258	self.cache_manager.put(new_url, contents, datetime.now() + self.caching_delta)	4✔
259
260	return contents	4✔
261
262	def fetch_style_from_local_path(self, partial_filename: str) -> Tuple[Optional[Path], str]:	4✔
263	"""Fetch a style file from a local path."""
264	if partial_filename and not partial_filename.endswith(TOML_EXTENSION):	4✔
265	partial_filename += TOML_EXTENSION	4✔
266	expanded_path = Path(partial_filename).expanduser()	4✔
267
268	if self._first_full_path and not (str(expanded_path).startswith("/") or partial_filename.startswith(DOT_SLASH)):	4✔
269	# Prepend the previous path to the partial file name.
270	style_path = Path(self._first_full_path) / expanded_path	4✔
271	else:
272	# Get the absolute path, be it from a root path (starting with slash) or from the current dir.
273	style_path = Path(expanded_path).absolute()	4✔
274
275	# Save the first full path to be used by the next files without parent.
276	if not self._first_full_path:	4✔
277	self._first_full_path = str(style_path.resolve().parent)	4✔
278
279	if str(style_path) in self._already_included:	4✔
280	return None, ""	4✔
281
282	if not style_path.exists():	4!
UNCOV 283	raise FileNotFoundError(f"Local style file does not exist: {style_path}")	×
284
285	logger.debug(f"Loading style from {style_path}")	4✔
286	self._already_included.add(str(style_path))	4✔
287	return style_path, style_path.read_text()	4✔
288
289	def merge_toml_dict(self) -> JsonDict:	4✔
290	"""Merge all included styles into a TOML (actually JSON) dictionary."""
291	merged_dict = self._all_styles.merge()	4✔
292	# TODO: check if the merged style file is still needed
293	merged_style_path: Path = self.cache_dir / MERGED_STYLE_TOML	4✔
294	toml = TOMLFormat(data=merged_dict)	4✔
295
296	attempt = 1	4✔
297	while attempt < 5:	4!
298	try:	4✔
299	merged_style_path.write_text(toml.reformatted)	4✔
300	break	4✔
UNCOV 301	except OSError:	×
UNCOV 302	attempt += 1	×
303
304	return merged_dict	4✔
305
306	@staticmethod	4✔
307	def file_field_pair(filename: str, base_file_class: Type[NitpickPlugin]) -> Dict[str, fields.Field]:	4✔
308	"""Return a schema field with info from a config file class."""
309	unique_filename_with_underscore = slugify(filename, separator="_")	4✔
310
311	kwargs = {"data_key": filename}	4✔
312	if base_file_class.validation_schema:	4!
UNCOV 313	field = fields.Nested(base_file_class.validation_schema, **kwargs)	×
314	else:
315	# For some files (e.g.: pyproject.toml, INI files), there is no strict schema;
316	# it can be anything they allow.
317	# It's out of Nitpick's scope to validate those files.
318	field = fields.Dict(fields.String, **kwargs)	4✔
319	return {unique_filename_with_underscore: field}	4✔
320
321	@lru_cache()	4✔
322	def load_fixed_dynamic_classes(self) -> Tuple[Plugins, Plugins]:	4✔
323	"""Separate classes with fixed file names from classes with dynamic files names."""
324	fixed_name_classes: Plugins = set()	4✔
325	dynamic_name_classes: Plugins = set()	4✔
326	for plugin_class in self.project.plugin_manager.hook.plugin_class(): # pylint: disable=no-member	4✔
327	if plugin_class.filename:	4✔
328	fixed_name_classes.add(plugin_class)	4✔
329	else:
330	dynamic_name_classes.add(plugin_class)	4✔
331	return fixed_name_classes, dynamic_name_classes	4✔
332
333	def rebuild_dynamic_schema(self, data: JsonDict = None) -> None:	4✔
334	"""Rebuild the dynamic Marshmallow schema when needed, adding new fields that were found on the style."""
335	new_files_found: Dict[str, fields.Field] = OrderedDict()	4✔
336
337	fixed_name_classes, dynamic_name_classes = self.load_fixed_dynamic_classes()	4✔
338
339	if data is None:	4!
340	# Data is empty; so this is the first time the dynamic class is being rebuilt.
341	# Loop on classes with predetermined names, and add fields for them on the dynamic validation schema.
342	# E.g.: pre-commit, pyproject.toml: files whose names we already know at this point.
343	for subclass in fixed_name_classes:	4✔
344	new_files_found.update(self.file_field_pair(subclass.filename, subclass))	4✔
345	else:
UNCOV 346	handled_tags: Dict[str, Type[NitpickPlugin]] = {}	×
347
348	# Data was provided; search it to find new dynamic files to add to the validation schema).
349	# E.g.: JSON files that were configured on some TOML style file.
UNCOV 350	for subclass in dynamic_name_classes:	×
UNCOV 351	for tag in subclass.identify_tags:	×
352	# TODO: WRONG! a tag should be handled by multiple classes...
353	# A tag can only be handled by a single subclass.
354	# If more than one class handle a tag, the latest one will be the handler.
UNCOV 355	handled_tags[tag] = subclass	×
356
UNCOV 357	jmex = subclass.get_compiled_jmespath_filenames()	×
358	for configured_filename in search_dict(jmex, data, []):	×
UNCOV 359	new_files_found.update(self.file_field_pair(configured_filename, subclass))	×
360
UNCOV 361	self._find_subclasses(data, handled_tags, new_files_found)	×
362
363	# Only recreate the schema if new fields were found.
364	if new_files_found:	4!
365	self._dynamic_schema_class = type("DynamicStyleSchema", (self._dynamic_schema_class,), new_files_found)	4✔
366
367	def _find_subclasses(self, data, handled_tags, new_files_found):	4✔
UNCOV 368	for possible_file in data.keys():	×
UNCOV 369	found_subclasses = []	×
UNCOV 370	for file_tag in identify.tags_from_filename(possible_file):	×
UNCOV 371	handler_subclass = handled_tags.get(file_tag)	×
UNCOV 372	if handler_subclass:	×
UNCOV 373	found_subclasses.append(handler_subclass)	×
374
UNCOV 375	for found_subclass in found_subclasses:	×
UNCOV 376	new_files_found.update(self.file_field_pair(possible_file, found_subclass))	×

andreoliwa / nitpick / 655228070

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