15808718976

Committed 22 Jun 2025 04:42PM UTC coverage: 96.64% (-0.1%) from 96.781%

Build # 15808718976

Build Type

Pull #119

github

Committed by

web-flow

Commit Message

Merge 0bc72d467 into afd15f6d2

Pull Request Pull Request #119: Feature/speed up ls

Run Details

4919 of 5090 relevant lines covered (96.64%)

7.73 hits per line

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

92.24

/src/ucdp/_modloader.py

#
# MIT License
#
# Copyright (c) 2024-2025 nbiotcloud
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#

"""
Loading And Searching Facility.
"""

import re
import sys
from collections.abc import Iterable, Iterator
from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache
from importlib import import_module
from inspect import getfile, isclass
from pathlib import Path
from typing import TypeAlias

from .consts import PKG_PATHS
from .modbase import BaseMod, get_modbaseclss
from .modref import ModRef, get_modclsname
from .modtopref import TopModRef
from .object import Object
from .util import LOGGER, get_maxworkers, guess_path

Patterns: TypeAlias = Iterable[str]
Paths: TypeAlias = Iterable[Path]

_RE_IMPORT_UCDP = re.compile(r"^\s*class .*Mod\):")

_RE_TOPMODREFPAT = re.compile(
    # [tb]#
    r"((?P<tb>[a-zA-Z_0-9_\.\*]+)#)?"
    # top
    r"(?P<top>[a-zA-Z_0-9_\.\*]+)"
    # [-sub]
    r"(-(?P<sub>[a-zA-Z_0-9_\.\*]+))?"
)


class TopModRefPat(Object):
    """Top Module Reference Search Pattern Pattern."""

    top: str
    sub: str | None = None
    tb: str | None = None

    def __str__(self):
        result = self.top
        if self.sub:
            result = f"{result}-{self.sub}"
        if self.tb:
            result = f"{self.tb}#{result}"
        return result


@lru_cache
def build_top(modcls, **kwargs):
    """Build Top Module."""
    return modcls.build_top(**kwargs)


@lru_cache
def load_modcls(modref: ModRef) -> type[BaseMod]:
    """Load Module Class."""
    name = f"{modref.libname}.{modref.modname}"
    try:
        pymod = import_module(name)
    except ModuleNotFoundError as exc:
        if exc.name in (modref.libname, name):
            raise NameError(f"{name!r} not found.") from None
        raise exc
    modclsname = modref.get_modclsname()
    modcls = getattr(pymod, modclsname, None)
    if not modcls:
        raise NameError(f"{name!r} does not contain {modclsname}.") from None
    if not issubclass(modcls, BaseMod):
        raise ValueError(f"{modcls} is not a module aka child of <class ucdp.BaseMod>.")
    return modcls


def find_modrefs(local: bool | None = None) -> tuple[ModRef, ...]:
    # determine directories with python files
    dirpaths: set[Path] = set()
    modrefs: list[ModRef] = []
    for syspathstr in sys.path:
        syspath = Path(syspathstr).resolve()
        if local is not None and local is any(syspath.is_relative_to(pkg_path) for pkg_path in PKG_PATHS):
            continue
        for filepath in syspath.glob("*/*.py"):
            dirpath = filepath.parent
            if dirpath.name.startswith("_") or dirpath.name == "ucdp":
                continue
            dirpaths.add(dirpath)

    maxworkers = get_maxworkers()
    with ThreadPoolExecutor(max_workers=maxworkers) as exe:
        # start
        jobs = [
            exe.submit(_find_modrefs, sys.path, tuple(sorted(dirpath.glob("*.py")))) for dirpath in sorted(dirpaths)
        ]
        # collect
        for job in jobs:
            modrefs.extend(job.result())
    return tuple(modrefs)


def _find_modrefs_files(modrefs: tuple[ModRef, ...], envpath: list[str], filepaths: tuple[Path, ...]) -> list[Path]:
    paths: set[Path] = set(filepaths)
    for modref in modrefs:
        modcls = load_modcls(modref)
        for basecls in get_modbaseclss(modcls):
            paths.add(Path(getfile(basecls)))
    return sorted(paths)


def _find_modrefs(envpath: list[str], filepaths: tuple[Path, ...]) -> tuple[ModRef, ...]:  # noqa: C901
    modrefs = []
    for filepath in filepaths:
        pylibname = filepath.parent.name
        pymodname = filepath.stem
        if pymodname.startswith("_"):
            continue
        # skip non-ucdp files
        try:
            with filepath.open(encoding="utf-8") as file:
                for line in file:
                    if _RE_IMPORT_UCDP.match(line):
                        break
                else:
                    continue
        except Exception as exc:
            LOGGER.info(f"Skipping {str(filepath)!r} ({exc})")
            continue

        # import module
        try:
            pymod = import_module(f"{pylibname}.{pymodname}")
        except Exception as exc:
            LOGGER.warning(f"Skipping {str(filepath)!r} ({exc})")
            continue

        # Inspect Module
        for name in dir(pymod):
            # Load Class
            modcls = getattr(pymod, name)
            if not isclass(modcls) or not issubclass(modcls, BaseMod):
                continue

            # Ignore imported
            if filepath != Path(getfile(modcls)):
                continue

            # Create ModRefInfo
            modclsname = get_modclsname(pymodname)
            if modclsname == name:
                modref = ModRef(libname=pylibname, modname=pymodname)
            else:
                modref = ModRef(libname=pylibname, modname=pymodname, modclsname=name)
            modrefs.append(modref)

    return tuple(modrefs)


def get_topmodrefpats(patterns: Patterns | None) -> Iterator[TopModRefPat | TopModRef]:
    for pattern in patterns or []:
        path = guess_path(pattern)
        if path:
            yield TopModRef.cast(path)
        else:
            mat = _RE_TOPMODREFPAT.fullmatch(pattern)
            if mat:
                yield TopModRefPat(**mat.groupdict())
            else:
                yield TopModRefPat(top=".")  # never matching

1	#
2	# MIT License
3	#
4	# Copyright (c) 2024-2025 nbiotcloud
5	#
6	# Permission is hereby granted, free of charge, to any person obtaining a copy
7	# of this software and associated documentation files (the "Software"), to deal
8	# in the Software without restriction, including without limitation the rights
9	# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10	# copies of the Software, and to permit persons to whom the Software is
11	# furnished to do so, subject to the following conditions:
12	#
13	# The above copyright notice and this permission notice shall be included in all
14	# copies or substantial portions of the Software.
15	#
16	# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17	# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18	# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19	# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20	# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21	# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22	# SOFTWARE.
23	#
24
25	"""
26	Loading And Searching Facility.
27	"""
28
29	import re	8✔
30	import sys	8✔
31	from collections.abc import Iterable, Iterator	8✔
32	from concurrent.futures import ThreadPoolExecutor	8✔
33	from functools import lru_cache	8✔
34	from importlib import import_module	8✔
35	from inspect import getfile, isclass	8✔
36	from pathlib import Path	8✔
37	from typing import TypeAlias	8✔
38
39	from .consts import PKG_PATHS	8✔
40	from .modbase import BaseMod, get_modbaseclss	8✔
41	from .modref import ModRef, get_modclsname	8✔
42	from .modtopref import TopModRef	8✔
43	from .object import Object	8✔
44	from .util import LOGGER, get_maxworkers, guess_path	8✔
45
46	Patterns: TypeAlias = Iterable[str]	8✔
47	Paths: TypeAlias = Iterable[Path]	8✔
48
49	_RE_IMPORT_UCDP = re.compile(r"^\sclass .Mod\):")	8✔
50
51	_RE_TOPMODREFPAT = re.compile(	8✔
52	# [tb]#
53	r"((?P<tb>[a-zA-Z_0-9_\.\*]+)#)?"
54	# top
55	r"(?P<top>[a-zA-Z_0-9_\.\*]+)"
56	# [-sub]
57	r"(-(?P<sub>[a-zA-Z_0-9_\.\*]+))?"
58	)
59
60
61	class TopModRefPat(Object):	8✔
62	"""Top Module Reference Search Pattern Pattern."""
63
64	top: str	8✔
65	sub: str \| None = None	8✔
66	tb: str \| None = None	8✔
67
68	def __str__(self):	8✔
69	result = self.top	8✔
70	if self.sub:	8✔
71	result = f"{result}-{self.sub}"	×
72	if self.tb:	8✔
73	result = f"{self.tb}#{result}"	×
74	return result	8✔
75
76
77	@lru_cache	8✔
78	def build_top(modcls, **kwargs):	8✔
79	"""Build Top Module."""
80	return modcls.build_top(**kwargs)	8✔
81
82
83	@lru_cache	8✔
84	def load_modcls(modref: ModRef) -> type[BaseMod]:	8✔
85	"""Load Module Class."""
86	name = f"{modref.libname}.{modref.modname}"	8✔
87	try:	8✔
88	pymod = import_module(name)	8✔
89	except ModuleNotFoundError as exc:	8✔
90	if exc.name in (modref.libname, name):	8✔
91	raise NameError(f"{name!r} not found.") from None	8✔
92	raise exc	8✔
93	modclsname = modref.get_modclsname()	8✔
94	modcls = getattr(pymod, modclsname, None)	8✔
95	if not modcls:	8✔
96	raise NameError(f"{name!r} does not contain {modclsname}.") from None	8✔
97	if not issubclass(modcls, BaseMod):	8✔
98	raise ValueError(f"{modcls} is not a module aka child of <class ucdp.BaseMod>.")	8✔
99	return modcls	8✔
100
101
102	def find_modrefs(local: bool \| None = None) -> tuple[ModRef, ...]:	8✔
103	# determine directories with python files
104	dirpaths: set[Path] = set()	8✔
105	modrefs: list[ModRef] = []	8✔
106	for syspathstr in sys.path:	8✔
107	syspath = Path(syspathstr).resolve()	8✔
108	if local is not None and local is any(syspath.is_relative_to(pkg_path) for pkg_path in PKG_PATHS):	8✔
109	continue	8✔
110	for filepath in syspath.glob("/.py"):	8✔
111	dirpath = filepath.parent	8✔
112	if dirpath.name.startswith("_") or dirpath.name == "ucdp":	8✔
113	continue	8✔
114	dirpaths.add(dirpath)	8✔
115
116	maxworkers = get_maxworkers()	8✔
117	with ThreadPoolExecutor(max_workers=maxworkers) as exe:	8✔
118	# start
119	jobs = [	8✔
120	exe.submit(_find_modrefs, sys.path, tuple(sorted(dirpath.glob("*.py")))) for dirpath in sorted(dirpaths)
121	]
122	# collect
123	for job in jobs:	8✔
124	modrefs.extend(job.result())	8✔
125	return tuple(modrefs)	8✔
126
127
128	def _find_modrefs_files(modrefs: tuple[ModRef, ...], envpath: list[str], filepaths: tuple[Path, ...]) -> list[Path]:	8✔
129	paths: set[Path] = set(filepaths)	×
130	for modref in modrefs:	×
131	modcls = load_modcls(modref)	×
132	for basecls in get_modbaseclss(modcls):	×
133	paths.add(Path(getfile(basecls)))	×
134	return sorted(paths)	×
135
136
137	def _find_modrefs(envpath: list[str], filepaths: tuple[Path, ...]) -> tuple[ModRef, ...]: # noqa: C901	8✔
138	modrefs = []	8✔
139	for filepath in filepaths:	8✔
140	pylibname = filepath.parent.name	8✔
141	pymodname = filepath.stem	8✔
142	if pymodname.startswith("_"):	8✔
143	continue	8✔
144	# skip non-ucdp files
145	try:	8✔
146	with filepath.open(encoding="utf-8") as file:	8✔
147	for line in file:	8✔
148	if _RE_IMPORT_UCDP.match(line):	8✔
149	break	8✔
150	else:
151	continue	8✔
152	except Exception as exc:	1✔
153	LOGGER.info(f"Skipping {str(filepath)!r} ({exc})")	1✔
154	continue	1✔
155
156	# import module
157	try:	8✔
158	pymod = import_module(f"{pylibname}.{pymodname}")	8✔
159	except Exception as exc:	8✔
160	LOGGER.warning(f"Skipping {str(filepath)!r} ({exc})")	8✔
161	continue	8✔
162
163	# Inspect Module
164	for name in dir(pymod):	8✔
165	# Load Class
166	modcls = getattr(pymod, name)	8✔
167	if not isclass(modcls) or not issubclass(modcls, BaseMod):	8✔
168	continue	8✔
169
170	# Ignore imported
171	if filepath != Path(getfile(modcls)):	8✔
172	continue	8✔
173
174	# Create ModRefInfo
175	modclsname = get_modclsname(pymodname)	8✔
176	if modclsname == name:	8✔
177	modref = ModRef(libname=pylibname, modname=pymodname)	8✔
178	else:
179	modref = ModRef(libname=pylibname, modname=pymodname, modclsname=name)	8✔
180	modrefs.append(modref)	8✔
181
182	return tuple(modrefs)	8✔
183
184
185	def get_topmodrefpats(patterns: Patterns \| None) -> Iterator[TopModRefPat \| TopModRef]:	8✔
186	for pattern in patterns or []:	8✔
187	path = guess_path(pattern)	8✔
188	if path:	8✔
189	yield TopModRef.cast(path)	×
190	else:
191	mat = _RE_TOPMODREFPAT.fullmatch(pattern)	8✔
192	if mat:	8✔
193	yield TopModRefPat(**mat.groupdict())	8✔
194	else:
195	yield TopModRefPat(top=".") # never matching	8✔

nbiotcloud / ucdp / 15808718976

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