tyrannosaurus.sync.Sync._replace() - Code Metrics - dmyersturnbull/tyrannosaurus - Measure and Improve Code Quality continuously with Scrutinizer

tyrannosaurus.sync.Sync._replace() A
last analyzed 2022-07-25 07:08 UTC

↳ Parent: tyrannosaurus.sync

Complexity

Conditions

Size

Total Lines	11
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	11
nop	4
dl	0
loc	11
rs	9.3333
c	0
b	0
f	0

"""
Module that syncs metadata from pyproject.toml.

Original source: https://github.com/dmyersturnbull/tyrannosaurus
Copyright 2020–2021 Douglas Myers-Turnbull
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at https://www.apache.org/licenses/LICENSE-2.0
"""
from __future__ import annotations

import logging
import re
import textwrap
from pathlib import Path
from typing import Mapping, Optional, Sequence, Union

from tyrannosaurus.context import Context
from tyrannosaurus.envs import CondaEnv

logger = logging.getLogger(__package__)


class Sync:
    def __init__(self, context: Context):
        self.context = context

    def sync(self) -> Sequence[str]:
        self.fix_init()
        self.fix_dockerfile()
        self.fix_pyproject()
        self.fix_recipe()
        self.fix_env()
        self.fix_codemeta()
        self.fix_citation()
        return [str(s) for s in self.context.targets]

    def has(self, key: str):
        return self.context.has_target(key)

    def fix_dockerfile(self) -> Sequence[str]:
        dockerfile = self.context.path / "Dockerfile"
        if not self.has("dockerfile") or not dockerfile.exists():
            return []
        oci_vr = "org.opencontainers.image.version"
        oci_desc = "org.opencontainers.image.description"
        vr = self.context.version
        desc = self.context.description
        return self._replace_substrs(
            dockerfile,
            {
                "LABEL version=": f'LABEL version="{vr}"',
                f"LABEL {oci_vr}=": f'LABEL {oci_vr}="{vr}"',
                f"LABEL {oci_desc}=": f'LABEL {oci_desc}="{desc}"',
            },
        )

    def fix_init(self) -> Sequence[str]:
        if self.has("init"):
            return self.fix_init_internal(self.context.path / self.context.project / "__init__.py")
        return []

    def fix_init_internal(self, init_path: Path) -> Sequence[str]:
        status = self.context.source("status")
        cright = self.context.source("copyright")
        dadate = self.context.source("date")
        return self._replace_substrs(
            init_path,
            {
                "__status__ = ": f'__status__ = "{status}"',
                "__copyright__ = ": f'__copyright__ = "{cright}"',
                "__date__ = ": f'__date__ = "{dadate}"',
            },
        )

    def fix_pyproject(self) -> Sequence[str]:
        if not self.has("pyproject"):
            return []
        version = self.context.version
        cz_version = self.context.data.get("tool.commitizen.version")
        version_from = self.context.data.get("tool.tyrannosaurus.sources.version")
        if cz_version is not None and cz_version != version:
            logger.error(f"Commitizen version {cz_version} != {version_from} version {version}")
        return []  # TODO: lying

    def fix_citation(self) -> Sequence[str]:
        path = self.context.path / "CITATION.cff"
        if not self.has("citation") or not path.exists():
            return []
        vr = self.context.version
        desc = self.context.description
        return self._replace_substrs(
            path,
            {
                "version:": f"version: {vr}",
                "^abstract:": f"abstract: {desc}",
            },
        )

    def fix_codemeta(self) -> Sequence[str]:
        path = self.context.path / "codemeta.json"
        if not self.has("codemeta") or not path.exists():
            return []
        vr = self.context.version
        desc = self.context.description
        return self._replace_substrs(
            path,
            {
                '    "version" *: *"': f'"version":"{vr}"',
                '    "description" *: *"': f'"description":"{desc}"',
            },
        )

    def fix_recipe(self) -> Sequence[str]:
        if (
            self.has("recipe")
            and self.context.path_source("recipe")
            and self.context.path_source("recipe").exists()
        ):
            return self.fix_recipe_internal(self.context.path_source("recipe"))
        return []

    def fix_env(self) -> Sequence[str]:
        if (
            self.has("environment")
            and self.context.path_source("environment")
            and self.context.path_source("environment").exists()
        ):
            creator = CondaEnv(self.context.project, dev=True, extras=True)
            return creator.create(self.context, self.context.path)
        return []

    def fix_recipe_internal(self, recipe_path: Path) -> Sequence[str]:
        # TODO this is all quite bad
        # Well, I guess this is still an alpha release
        # python_vr = self.context.deps["python"]
        pat = re.compile(r"github:([a-z\d](?:[a-z\d]|-(?=[a-z\d])){0,38})")
        summary = self._careful_wrap(self.context.poetry("description"))
        if "long_description" in self.context.sources:
            long_desc = self._careful_wrap(self.context.source("long_description"))
        else:
            long_desc = summary
        poetry_vr = self.context.build_sys_reqs["poetry"]
        maintainers = self.context.source("maintainers")
        maintainers = [m.group(1) for m in pat.finditer(maintainers)]
        maintainers = "\n    - ".join(maintainers)
        # the pip >= 20 gets changed for BOTH test and host; this is OK
        # The same is true for the poetry >=1.1,<2.0 line: it's added to both sections
        vr_strp = poetry_vr.replace(" ", "")
        lines = self._replace_substrs(
            recipe_path,
            {
                "{% set version = ": '{% set version = "' + str(self.context.version) + '" %}',
                "    - python >=": f"    - python {vr_strp}",
                re.compile("^ {4}- pip *$"): f"    - pip >=20\n    - poetry {vr_strp}",
            },
        )
        new_lines = self._until_line(lines, "about:")
        last_section = f"""
about:
  home: {self.context.poetry("homepage")}
  summary: |
    {summary}
  license_family: {self.context.license.family}
  license: {self.context.license.spdx}
  license_file: LICENSE.txt
  description: |
    {long_desc}
  doc_url: {self.context.poetry("documentation")}
  dev_url: {self.context.poetry("repository")}

extra:
  recipe-maintainers:
    - {maintainers}
"""
        final_lines = [*new_lines, *last_section.splitlines()]
        final_lines = [x.rstrip(" ") for x in final_lines]
        final_str = "\n".join(final_lines)
        final_str = re.compile(r"\n\s*\n").sub("\n\n", final_str)
        if not self.context.dry_run:
            recipe_path.write_text(final_str, encoding="utf8")
        logger.debug(f"Wrote to {recipe_path}")
        return final_str.split("\n")

    def _until_line(self, lines: Sequence[str], stop_at: str):
        new_lines = []
        for line in lines:
            if line.startswith(stop_at):
                break
            new_lines.append(line)
        return new_lines

    def _careful_wrap(self, s: str, indent: int = 4) -> str:
        txt = " ".join(s.split())
        width = self._get_line_length()
        # TODO: I don't know why replace_whitespace=True, drop_whitespace=True isn't sufficient
        return textwrap.fill(
            txt,
            width=width,
            subsequent_indent=" " * indent,
            break_long_words=False,
            break_on_hyphens=False,
            replace_whitespace=True,
            drop_whitespace=True,
        )

    def _replace_substrs(
        self,
        path: Path,
        replace: Mapping[Union[str, re.Pattern], str],
    ) -> Sequence[str]:
        if not self.context.dry_run:
            self.context.back_up(path)
        new_lines = "\n".join(
            [self._fix_line(line, replace) for line in path.read_text(encoding="utf8").splitlines()]
        )
        if not self.context.dry_run:
            path.write_text(new_lines, encoding="utf8")
        logger.debug(f"Wrote to {path}")
        return new_lines.splitlines()

    def _fix_line(self, line: str, replace: Mapping[Union[str, re.Pattern], str]) -> str:
        for k, v in replace.items():
            replace = self._replace(line, k, v)
            if replace is not None:
                return replace
        else:
            return line

    def _replace(self, line: str, k: Union[str, re.Pattern], v: str) -> Optional[str]:
        if isinstance(k, re.Pattern):
            try:
                if k.fullmatch(line) is not None:
                    return k.sub(line, v)
            except re.error:
                logger.error(f"Failed to process '{line}' with pattern '{k}")
                raise
        elif line.startswith(k):
            return v
        return None

    def _get_line_length(self) -> int:
        if "linelength" in self.context.sources:
            return int(self.context.source("linelength"))
        elif "tool.black.line-length" in self.context.data:
            return int(self.context.data["tool.black.line-length"])
        return 100


__all__ = ["Sync"]


1			"""
2			Module that syncs metadata from pyproject.toml.
3
4			Original source: https://github.com/dmyersturnbull/tyrannosaurus
5			Copyright 2020–2021 Douglas Myers-Turnbull
6			Licensed under the Apache License, Version 2.0 (the "License");
7			you may not use this file except in compliance with the License.
8			You may obtain a copy of the License at https://www.apache.org/licenses/LICENSE-2.0
9			"""
10			from __future__ import annotations
11
12			import logging
13			import re
14			import textwrap
15			from pathlib import Path
16			from typing import Mapping, Optional, Sequence, Union
17
18			from tyrannosaurus.context import Context
19			from tyrannosaurus.envs import CondaEnv
20
21			logger = logging.getLogger(__package__)
22
23
24			class Sync:
25			def __init__(self, context: Context):
26			self.context = context
27
28			def sync(self) -> Sequence[str]:
29			self.fix_init()
30			self.fix_dockerfile()
31			self.fix_pyproject()
32			self.fix_recipe()
33			self.fix_env()
34			self.fix_codemeta()
35			self.fix_citation()
36			return [str(s) for s in self.context.targets]
37
38			def has(self, key: str):
39			return self.context.has_target(key)
40
41			def fix_dockerfile(self) -> Sequence[str]:
42			dockerfile = self.context.path / "Dockerfile"
43			if not self.has("dockerfile") or not dockerfile.exists():
44			return []
45			oci_vr = "org.opencontainers.image.version"
46			oci_desc = "org.opencontainers.image.description"
47			vr = self.context.version
48			desc = self.context.description
49			return self._replace_substrs(
50			dockerfile,
51			{
52			"LABEL version=": f'LABEL version="{vr}"',
53			f"LABEL {oci_vr}=": f'LABEL {oci_vr}="{vr}"',
54			f"LABEL {oci_desc}=": f'LABEL {oci_desc}="{desc}"',
55			},
56			)
57
58			def fix_init(self) -> Sequence[str]:
59			if self.has("init"):
60			return self.fix_init_internal(self.context.path / self.context.project / "__init__.py")
61			return []
62
63			def fix_init_internal(self, init_path: Path) -> Sequence[str]:
64			status = self.context.source("status")
65			cright = self.context.source("copyright")
66			dadate = self.context.source("date")
67			return self._replace_substrs(
68			init_path,
69			{
70			"__status__ = ": f'__status__ = "{status}"',
71			"__copyright__ = ": f'__copyright__ = "{cright}"',
72			"__date__ = ": f'__date__ = "{dadate}"',
73			},
74			)
75
76			def fix_pyproject(self) -> Sequence[str]:
77			if not self.has("pyproject"):
78			return []
79			version = self.context.version
80			cz_version = self.context.data.get("tool.commitizen.version")
81			version_from = self.context.data.get("tool.tyrannosaurus.sources.version")
82			if cz_version is not None and cz_version != version:
83			logger.error(f"Commitizen version {cz_version} != {version_from} version {version}")
84			return [] # TODO: lying
85
86			def fix_citation(self) -> Sequence[str]:
87			path = self.context.path / "CITATION.cff"
88			if not self.has("citation") or not path.exists():
89			return []
90			vr = self.context.version
91			desc = self.context.description
92			return self._replace_substrs(
93			path,
94			{
95			"version:": f"version: {vr}",
96			"^abstract:": f"abstract: {desc}",
97			},
98			)
99
100			def fix_codemeta(self) -> Sequence[str]:
101			path = self.context.path / "codemeta.json"
102			if not self.has("codemeta") or not path.exists():
103			return []
104			vr = self.context.version
105			desc = self.context.description
106			return self._replace_substrs(
107			path,
108			{
109			' "version" : "': f'"version":"{vr}"',
110			' "description" : "': f'"description":"{desc}"',
111			},
112			)
113
114			def fix_recipe(self) -> Sequence[str]:
115			if (
116			self.has("recipe")
117			and self.context.path_source("recipe")
118			and self.context.path_source("recipe").exists()
119			):
120			return self.fix_recipe_internal(self.context.path_source("recipe"))
121			return []
122
123			def fix_env(self) -> Sequence[str]:
124			if (
125			self.has("environment")
126			and self.context.path_source("environment")
127			and self.context.path_source("environment").exists()
128			):
129			creator = CondaEnv(self.context.project, dev=True, extras=True)
130			return creator.create(self.context, self.context.path)
131			return []
132
133			def fix_recipe_internal(self, recipe_path: Path) -> Sequence[str]:
134			# TODO this is all quite bad
135			# Well, I guess this is still an alpha release
136			# python_vr = self.context.deps["python"]
137			pat = re.compile(r"github:([a-z\d](?:[a-z\d]\|-(?=[a-z\d])){0,38})")
138			summary = self._careful_wrap(self.context.poetry("description"))
139			if "long_description" in self.context.sources:
140			long_desc = self._careful_wrap(self.context.source("long_description"))
141			else:
142			long_desc = summary
143			poetry_vr = self.context.build_sys_reqs["poetry"]
144			maintainers = self.context.source("maintainers")
145			maintainers = [m.group(1) for m in pat.finditer(maintainers)]
146			maintainers = "\n - ".join(maintainers)
147			# the pip >= 20 gets changed for BOTH test and host; this is OK
148			# The same is true for the poetry >=1.1,<2.0 line: it's added to both sections
149			vr_strp = poetry_vr.replace(" ", "")
150			lines = self._replace_substrs(
151			recipe_path,
152			{
153			"{% set version = ": '{% set version = "' + str(self.context.version) + '" %}',
154			" - python >=": f" - python {vr_strp}",
155			re.compile("^ {4}- pip *$"): f" - pip >=20\n - poetry {vr_strp}",
156			},
157			)
158			new_lines = self._until_line(lines, "about:")
159			last_section = f"""
160			about:
161			home: {self.context.poetry("homepage")}
162			summary: \|
163			{summary}
164			license_family: {self.context.license.family}
165			license: {self.context.license.spdx}
166			license_file: LICENSE.txt
167			description: \|
168			{long_desc}
169			doc_url: {self.context.poetry("documentation")}
170			dev_url: {self.context.poetry("repository")}
171
172			extra:
173			recipe-maintainers:
174			- {maintainers}
175			"""
176			final_lines = [new_lines, last_section.splitlines()]
177			final_lines = [x.rstrip(" ") for x in final_lines]
178			final_str = "\n".join(final_lines)
179			final_str = re.compile(r"\n\s*\n").sub("\n\n", final_str)
180			if not self.context.dry_run:
181			recipe_path.write_text(final_str, encoding="utf8")
182			logger.debug(f"Wrote to {recipe_path}")
183			return final_str.split("\n")
184
185			def _until_line(self, lines: Sequence[str], stop_at: str):
186			new_lines = []
187			for line in lines:
188			if line.startswith(stop_at):
189			break
190			new_lines.append(line)
191			return new_lines
192
193			def _careful_wrap(self, s: str, indent: int = 4) -> str:
194			txt = " ".join(s.split())
195			width = self._get_line_length()
196			# TODO: I don't know why replace_whitespace=True, drop_whitespace=True isn't sufficient
197			return textwrap.fill(
198			txt,
199			width=width,
200			subsequent_indent=" " * indent,
201			break_long_words=False,
202			break_on_hyphens=False,
203			replace_whitespace=True,
204			drop_whitespace=True,
205			)
206
207			def _replace_substrs(
208			self,
209			path: Path,
210			replace: Mapping[Union[str, re.Pattern], str],
211			) -> Sequence[str]:
212			if not self.context.dry_run:
213			self.context.back_up(path)
214			new_lines = "\n".join(
215			[self._fix_line(line, replace) for line in path.read_text(encoding="utf8").splitlines()]
216			)
217			if not self.context.dry_run:
218			path.write_text(new_lines, encoding="utf8")
219			logger.debug(f"Wrote to {path}")
220			return new_lines.splitlines()
221
222			def _fix_line(self, line: str, replace: Mapping[Union[str, re.Pattern], str]) -> str:
223			for k, v in replace.items():
224			replace = self._replace(line, k, v)
225			if replace is not None:
226			return replace
227			else:
228			return line
229
230			def _replace(self, line: str, k: Union[str, re.Pattern], v: str) -> Optional[str]:
231			if isinstance(k, re.Pattern):
232			try:
233			if k.fullmatch(line) is not None:
234			return k.sub(line, v)
235			except re.error:
236			logger.error(f"Failed to process '{line}' with pattern '{k}")
237			raise
238			elif line.startswith(k):
239			return v
240			return None
241
242			def _get_line_length(self) -> int:
243			if "linelength" in self.context.sources:
244			return int(self.context.source("linelength"))
245			elif "tool.black.line-length" in self.context.data:
246			return int(self.context.data["tool.black.line-length"])
247			return 100
248
249
250			__all__ = ["Sync"]
251

dmyersturnbull / tyrannosaurus

tyrannosaurus.sync.Sync._replace() A last analyzed 2022-07-25 07:08 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

tyrannosaurus.sync.Sync._replace() A
last analyzed 2022-07-25 07:08 UTC