1import functools
2import logging
3import sys
4import time
5import types
6
7if sys.version_info >= (3, 10):
8    # Python 3.10 will include a fix in importlib.metadata which allows us to
9    # get the distribution of a loaded entry-point
10    import importlib.metadata  # pylint: disable=no-member,no-name-in-module
11
12    USE_IMPORTLIB_METADATA_STDLIB = True
13else:
14    USE_IMPORTLIB_METADATA_STDLIB = False
15    try:
16        from salt._compat import importlib_metadata
17
18        USE_IMPORTLIB_METADATA = True
19    except ImportError:
20        USE_IMPORTLIB_METADATA = False
21
22log = logging.getLogger(__name__)
23
24
25def timed_lru_cache(timeout_seconds, *, maxsize=256, typed=False):
26    """
27    This decorator is the same in behavior as functools.lru_cache with the
28    exception that it times out after the provided ``timeout_seconds``
29    """
30
31    def _wrapper(f):
32        # Apply @lru_cache to f
33        f = functools.lru_cache(maxsize=maxsize, typed=typed)(f)
34        f.delta = timeout_seconds
35        f.expiration = time.monotonic() + f.delta
36
37        @functools.wraps(f)
38        def _wrapped(*args, **kwargs):
39            now = time.monotonic()
40            if now >= f.expiration:
41                f.cache_clear()
42                f.expiration = now + f.delta
43            return f(*args, **kwargs)
44
45        return _wrapped
46
47    return _wrapper
48
49
50@timed_lru_cache(timeout_seconds=0.5)
51def iter_entry_points(group, name=None):
52    entry_points_listing = []
53    if USE_IMPORTLIB_METADATA_STDLIB:
54        log.debug("Using importlib.metadata to load entry points")
55        entry_points = importlib.metadata.entry_points()
56    elif USE_IMPORTLIB_METADATA:
57        log.debug("Using importlib_metadata to load entry points")
58        entry_points = importlib_metadata.entry_points()
59    else:
60        return entry_points_listing
61
62    for entry_point_group, entry_points_list in entry_points.items():
63        if entry_point_group != group:
64            continue
65        for entry_point in entry_points_list:
66            if name is not None and entry_point.name != name:
67                continue
68            entry_points_listing.append(entry_point)
69
70    return entry_points_listing
71
72
73def name_and_version_from_entry_point(entry_point):
74    return types.SimpleNamespace(
75        name=entry_point.dist.metadata["name"],
76        version=entry_point.dist.version,
77    )
78