1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
from __future__ import annotations
import shutil
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from hashlib import sha1
from pathlib import Path
from typing import Any, Dict, NamedTuple
RUFF_BASE_ARGS = [
"ruff",
"--no-cache",
"--fix",
"--quiet",
"--select",
"UP",
"--select",
"F401",
"--isolated",
"-",
"--target-version",
]
class VersionTuple(NamedTuple):
major: int
minor: int
@classmethod
def from_string(cls, version: str) -> VersionTuple:
major, minor = version.split(".")
return VersionTuple(major=int(major), minor=int(minor))
VersionedCode = Dict[VersionTuple, str]
class Cache:
"""Simple hybrid file system / memory cache.
Follows the
`Cache Directory Tagging Specification http://www.brynosaurus.com/cachedir/>`_.
"""
def __init__(self) -> None:
self.cache_dir = Path(".auto_pytabs_cache")
self.cache_content_dir = self.cache_dir / "content"
self._cache: dict[str, str] = {}
self._touched: set[str] = set()
self._load()
def _init_cache_dir(self) -> None:
self.cache_content_dir.mkdir(exist_ok=True, parents=True)
cachedir_tag = self.cache_dir / "CACHEDIR.TAG"
gitignore_file = self.cache_dir / ".gitignore"
if not cachedir_tag.exists():
cachedir_tag.write_text(
"""Signature: 8a477f597d28d172789f06886806bc55
# This file is a cache directory tag created by (application name).
# For information about cache directory tags, see:
# http://www.brynosaurus.com/cachedir/"""
)
if not gitignore_file.exists():
gitignore_file.write_text("*\n")
def _load(self) -> None:
self._init_cache_dir()
cache: dict[str, str] = {}
with ThreadPoolExecutor() as executor:
futures = {
executor.submit(file.read_text): file.name
for file in self.cache_content_dir.iterdir()
}
for future in as_completed(futures):
cache[futures[future]] = future.result()
self._cache.update(cache)
@staticmethod
def make_cache_key(*parts: Any) -> str:
"""Create a cache key using an md5 hash of ``parts``."""
return sha1("".join(map(str, parts)).encode()).hexdigest()
def get(self, key: str) -> str | None:
"""Get an item specified by ``key`` the cache."""
self._touched.add(key)
return self._cache.get(key)
def set(self, key: str, content: str) -> None:
"""Store an ``content``."""
self._cache[key] = content
self._touched.add(key)
def persist(self, evict: bool = True) -> None:
"""
Persist internal cache to disk. If ``evict`` is ``True``, evict unused items.
"""
with ThreadPoolExecutor() as executor:
for key, content in self._cache.items():
if key in self._touched:
executor.submit(
self.cache_content_dir.joinpath(key).write_text, content
)
elif evict:
executor.submit(
self.cache_content_dir.joinpath(key).unlink, missing_ok=True
)
def clear_all(self) -> None:
"""Clear all cached items from memory and disk."""
self._cache = {}
self._touched = set()
if not self.cache_dir.exists():
return
shutil.rmtree(self.cache_dir)
def get_version_requirements(
min_version: VersionTuple, max_version: VersionTuple
) -> list[VersionTuple]:
"""Given a min and max version, generate all versions in between."""
min_major, min_minor = min_version
max_major, max_minor = max_version
return [
VersionTuple(major=major, minor=minor)
for major in range(min_major, max_major + 1)
for minor in range(min_minor, max_minor + 1)
]
def _run_ruff(source: str, target_version: str) -> str:
with subprocess.Popen(
[*RUFF_BASE_ARGS, target_version],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.PIPE,
encoding="utf-8",
) as process:
return process.communicate(input=source)[0]
def _upgrade_code(
code: str, min_version: VersionTuple, cache: Cache | None = None
) -> str:
cache_key: str | None = None
if cache:
cache_key = cache.make_cache_key(code, min_version)
if cached_code := cache.get(cache_key):
return cached_code
code = _run_ruff(code, target_version=f"py3{min_version.minor}")
if cache_key and cache:
cache.set(cache_key, code)
return code
def version_code(
code: str, version_requirements: list[VersionTuple], cache: Cache | None = None
) -> VersionedCode:
"""Create versions of ``code`` for all python versions specified in
``version_requirements`` and return a dictionary of version-tuples/code.
"""
latest_code = code
versioned_code: VersionedCode = {version_requirements[0]: code}
for version in version_requirements:
upgraded_code = _upgrade_code(latest_code, version, cache=cache)
if upgraded_code != latest_code:
versioned_code[version] = upgraded_code
latest_code = upgraded_code
return versioned_code
|
