18093583429

Committed 29 Sep 2025 10:16AM UTC coverage: 90.562% (-6.1%) from 96.647%

Build # 18093583429

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #132 from nbiotcloud/update-deps

update dependencies

Run Details

4673 of 5160 relevant lines covered (90.56%)

10.86 hits per line

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

87.18

/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 .pathutil import absolute
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 = absolute(Path(syspathstr))
        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	12✔
30	import sys	12✔
31	from collections.abc import Iterable, Iterator	12✔
32	from concurrent.futures import ThreadPoolExecutor	12✔
33	from functools import lru_cache	12✔
34	from importlib import import_module	12✔
35	from inspect import getfile, isclass	12✔
36	from pathlib import Path	12✔
37	from typing import TypeAlias	12✔
38
39	from .consts import PKG_PATHS	12✔
40	from .modbase import BaseMod, get_modbaseclss	12✔
41	from .modref import ModRef, get_modclsname	12✔
42	from .modtopref import TopModRef	12✔
43	from .object import Object	12✔
44	from .pathutil import absolute	12✔
45	from .util import LOGGER, get_maxworkers, guess_path	12✔
46
47	Patterns: TypeAlias = Iterable[str]	12✔
48	Paths: TypeAlias = Iterable[Path]	12✔
49
50	_RE_IMPORT_UCDP = re.compile(r"^\sclass .Mod\):")	12✔
51
52	_RE_TOPMODREFPAT = re.compile(	12✔
53	# [tb]#
54	r"((?P<tb>[a-zA-Z_0-9_\.\*]+)#)?"
55	# top
56	r"(?P<top>[a-zA-Z_0-9_\.\*]+)"
57	# [-sub]
58	r"(-(?P<sub>[a-zA-Z_0-9_\.\*]+))?"
59	)
60
61
62	class TopModRefPat(Object):	12✔
63	"""Top Module Reference Search Pattern Pattern."""
64
65	top: str	12✔
66	sub: str \| None = None	12✔
67	tb: str \| None = None	12✔
68
69	def __str__(self):	12✔
70	result = self.top	×
71	if self.sub:	×
72	result = f"{result}-{self.sub}"	×
73	if self.tb:	×
74	result = f"{self.tb}#{result}"	×
75	return result	×
76
77
78	@lru_cache	12✔
79	def build_top(modcls, **kwargs):	12✔
80	"""Build Top Module."""
81	return modcls.build_top(**kwargs)	12✔
82
83
84	@lru_cache	12✔
85	def load_modcls(modref: ModRef) -> type[BaseMod]:	12✔
86	"""Load Module Class."""
87	name = f"{modref.libname}.{modref.modname}"	12✔
88	try:	12✔
89	pymod = import_module(name)	12✔
90	except ModuleNotFoundError as exc:	12✔
91	if exc.name in (modref.libname, name):	12✔
92	raise NameError(f"{name!r} not found.") from None	12✔
93	raise exc	12✔
94	modclsname = modref.get_modclsname()	12✔
95	modcls = getattr(pymod, modclsname, None)	12✔
96	if not modcls:	12✔
97	raise NameError(f"{name!r} does not contain {modclsname}.") from None	12✔
98	if not issubclass(modcls, BaseMod):	12✔
99	raise ValueError(f"{modcls} is not a module aka child of <class ucdp.BaseMod>.")	12✔
100	return modcls	12✔
101
102
103	def find_modrefs(local: bool \| None = None) -> tuple[ModRef, ...]:	12✔
104	# determine directories with python files
105	dirpaths: set[Path] = set()	12✔
106	modrefs: list[ModRef] = []	12✔
107	for syspathstr in sys.path:	12✔
108	syspath = absolute(Path(syspathstr))	12✔
109	if local is not None and local is any(syspath.is_relative_to(pkg_path) for pkg_path in PKG_PATHS):	12✔
110	continue	×
111	for filepath in syspath.glob("/.py"):	12✔
112	dirpath = filepath.parent	12✔
113	if dirpath.name.startswith("_") or dirpath.name == "ucdp":	12✔
114	continue	12✔
115	dirpaths.add(dirpath)	12✔
116
117	maxworkers = get_maxworkers()	12✔
118	with ThreadPoolExecutor(max_workers=maxworkers) as exe:	12✔
119	# start
120	jobs = [	12✔
121	exe.submit(_find_modrefs, sys.path, tuple(sorted(dirpath.glob("*.py")))) for dirpath in sorted(dirpaths)
122	]
123	# collect
124	for job in jobs:	12✔
125	modrefs.extend(job.result())	12✔
126	return tuple(modrefs)	12✔
127
128
129	def _find_modrefs_files(modrefs: tuple[ModRef, ...], envpath: list[str], filepaths: tuple[Path, ...]) -> list[Path]:	12✔
130	paths: set[Path] = set(filepaths)	×
131	for modref in modrefs:	×
132	modcls = load_modcls(modref)	×
133	for basecls in get_modbaseclss(modcls):	×
134	paths.add(Path(getfile(basecls)))	×
135	return sorted(paths)	×
136
137
138	def _find_modrefs(envpath: list[str], filepaths: tuple[Path, ...]) -> tuple[ModRef, ...]: # noqa: C901	12✔
139	modrefs = []	12✔
140	for filepath in filepaths:	12✔
141	pylibname = filepath.parent.name	12✔
142	pymodname = filepath.stem	12✔
143	if pymodname.startswith("_"):	12✔
144	continue	12✔
145	# skip non-ucdp files
146	try:	12✔
147	with filepath.open(encoding="utf-8") as file:	12✔
148	for line in file:	12✔
149	if _RE_IMPORT_UCDP.match(line):	12✔
150	break	12✔
151	else:
152	continue	12✔
153	except Exception as exc:	3✔
154	LOGGER.info(f"Skipping {str(filepath)!r} ({exc})")	3✔
155	continue	3✔
156
157	# import module
158	try:	12✔
159	pymod = import_module(f"{pylibname}.{pymodname}")	12✔
160	except Exception as exc:	12✔
161	LOGGER.warning(f"Skipping {str(filepath)!r} ({exc})")	12✔
162	continue	12✔
163
164	# Inspect Module
165	for name in dir(pymod):	12✔
166	# Load Class
167	modcls = getattr(pymod, name)	12✔
168	if not isclass(modcls) or not issubclass(modcls, BaseMod):	12✔
169	continue	12✔
170
171	# Ignore imported
172	if filepath != Path(getfile(modcls)):	12✔
173	continue	12✔
174
175	# Create ModRefInfo
176	modclsname = get_modclsname(pymodname)	12✔
177	if modclsname == name:	12✔
178	modref = ModRef(libname=pylibname, modname=pymodname)	12✔
179	else:
180	modref = ModRef(libname=pylibname, modname=pymodname, modclsname=name)	12✔
181	modrefs.append(modref)	12✔
182
183	return tuple(modrefs)	12✔
184
185
186	def get_topmodrefpats(patterns: Patterns \| None) -> Iterator[TopModRefPat \| TopModRef]:	12✔
187	for pattern in patterns or []:	12✔
188	path = guess_path(pattern)	12✔
189	if path:	12✔
190	yield TopModRef.cast(path)	×
191	else:
192	mat = _RE_TOPMODREFPAT.fullmatch(pattern)	12✔
193	if mat:	12✔
194	yield TopModRefPat(**mat.groupdict())	12✔
195	else:
196	yield TopModRefPat(top=".") # never matching	×

nbiotcloud / ucdp / 18093583429

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