Mercurial > hgrepos > Python > apps > py-cutils

comparison cutils/util/walk.py @ 178:dac26a2d9de5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Cleanup: remove non used walk-code
author Franz Glasner <fzglas.hg@dom66.de>
date Sat, 11 Jan 2025 21:18:53 +0100
parents 089c40240061
children 188de62caac6
comparison
equal deleted inserted replaced
177:089c40240061 178:dac26a2d9de5
5 # :- 5 # :-
6 r"""Utility sub-module to implement a heavily customized :func:`os.walk`. 6 r"""Utility sub-module to implement a heavily customized :func:`os.walk`.
7 7
8 """ 8 """
9 9
10 __all__ = ["walk", 10 __all__ = ["ScanDir"]
11 "ScanDir"]
12 11
13 12
14 import os 13 import os
15 try: 14 try:
16 from os import scandir 15 from os import scandir
19 from scandir import scandir 18 from scandir import scandir
20 except ImportError: 19 except ImportError:
21 scandir = None 20 scandir = None
22 21
23 from . import PY2 22 from . import PY2
24 from .cm import nullcontext
25 23
26 24
27 class WalkDirEntry(object): 25 class WalkDirEntry(object):
28 26
29 """A :class:`os.DirEntry` alike to be used in :func:`walk` and for 27 """A :class:`os.DirEntry` alike to be used in :func:`walk` and for
140 @staticmethod 138 @staticmethod
141 def sort_key(entry): 139 def sort_key(entry):
142 return entry._fsname 140 return entry._fsname
143 141
144 142
145 def walk(root, follow_symlinks=False):
146 """A heyvily customized :func:`os.walk` alike that differs from the
147 original:
148
149 - optimized for use in :command:`treesum`
150 - most errors are not suppressed
151 - the `root` is never part of the returned data
152 - the returned directory in "top" is not a string form but a list of
153 individual path segments
154 - there is only one yielded list
155
156 * contains :class:`WalkDirEntry`
157 * sorted by its fsname
158
159 The caller can easily get the old dirs and nondirs by filtering
160 the yielded list using "entry.is_dir".
161
162 - recurse into sub-directories first ("topdown=False")
163 - sort consistently all yielded lists by the filesystem encoding
164
165 .. note:: The implementation is based on Python 3.11 and needs a
166 functional :func:`os.scandir` or :func:`scandir.scandir`
167 implementation. It intentionally follows the logic in
168 Python 3.11 while it could be simplified because we are not
169 implementing some of the original flags (e.g. like
170 `topdown`).
171
172 """
173 normed_root = os.path.normpath(root)
174 yield from _walk(normed_root, tuple(), follow_symlinks=follow_symlinks)
175
176
177 if scandir: 143 if scandir:
178 144
179 def _walk(root, top, follow_symlinks): 145 class ScanDir(object):
180 """:func:`walk` helper.
181
182 Implemented using :func:`os.scandir`.
183
184 """
185 if top:
186 path = os.path.join(root, *top)
187 else:
188 path = root
189
190 fsobjects, walk_dirs = [], []
191
192 scandir_cm = scandir(path)
193 if not hasattr(scandir_cm, "close"):
194 scandir_cm = nullcontext(scandir_cm)
195 with scandir_cm as scandir_it:
196 while True:
197 try:
198 entry = WalkDirEntry.from_direntry(next(scandir_it))
199 except StopIteration:
200 break
201 fsobjects.append(entry)
202 #
203 # Always bottom-up: recurse into sub-directories, but exclude
204 # symlinks to directories if follow_symlinks is False
205 #
206 if entry.is_dir:
207 if follow_symlinks:
208 walk_into = True
209 else:
210 walk_into = not entry.is_symlink
211 if walk_into:
212 walk_dirs.append(entry)
213
214 # Sort by low-level filesystem encoding
215 walk_dirs.sort(key=WalkDirEntry.sort_key)
216 fsobjects.sort(key=WalkDirEntry.sort_key)
217
218 # Recurse into sub-directories
219 for wd in walk_dirs:
220 yield from _walk(root, top + (wd.name,), follow_symlinks)
221 # Yield after recursion if going bottom up
222 yield top, fsobjects
223
224
225 class ScanDir(object): # noqa: E303 too many blank lines
226 146
227 """An :func:`os.scandir` wrapper that is always an iterator and 147 """An :func:`os.scandir` wrapper that is always an iterator and
228 a context manager. 148 a context manager.
229 149
230 """ 150 """
251 if hasattr(self._scandir_it, "close"): 171 if hasattr(self._scandir_it, "close"):
252 self._scandir_it.close() 172 self._scandir_it.close()
253 173
254 else: 174 else:
255 175
256 def _walk(root, top, follow_symlinks): 176 class ScanDir(object):
257 """:func:`walk` helper.
258
259 Implemented using :func:`os.listdir`.
260
261 """
262 if top:
263 path = os.path.join(root, *top)
264 else:
265 path = root
266
267 fsobjects, walk_dirs = [], []
268
269 names = os.listdir(path)
270 for name in names:
271 entry = WalkDirEntry.from_path_name(path, name)
272 fsobjects.append(entry)
273 #
274 # Always bottom-up: recurse into sub-directories, but exclude
275 # symlinks to directories if follow_symlinks is False
276 #
277 if entry.is_dir:
278 if follow_symlinks:
279 walk_into = True
280 else:
281 walk_into = not entry.is_symlink
282 if walk_into:
283 walk_dirs.append(entry)
284
285 # Sort by low-level filesystem encoding
286 walk_dirs.sort(key=WalkDirEntry.sort_key)
287 fsobjects.sort(key=WalkDirEntry.sort_key)
288
289 # Recurse into sub-directories
290 for wd in walk_dirs:
291 yield from _walk(root, top + (wd.name,), follow_symlinks)
292 # Yield after recursion if going bottom up
293 yield top, fsobjects
294
295
296 class ScanDir(object): # noqa: E303 too many blank lines
297 177
298 """An :func:`os.scandir` wrapper that is always an iterator and 178 """An :func:`os.scandir` wrapper that is always an iterator and
299 a context manager. 179 a context manager.
300 180
301 """ 181 """