/
driver_cache.py
60 lines (46 loc) · 2.01 KB
/
driver_cache.py
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
# This file is part of the Open Data Cube, see https://opendatacube.org for more information
#
# Copyright (c) 2015-2023 ODC Contributors
# SPDX-License-Identifier: Apache-2.0
import logging
from typing import Dict, Any, Tuple, Iterable
_LOG = logging.getLogger(__name__)
def load_drivers(group: str) -> Dict[str, Any]:
"""
Load available drivers for a given group name.
Gracefully handles:
- Driver module not able to be imported
- Driver init function throwing an exception or returning None
By having driver entry_points pointing to a function, we defer loading the driver
module or running any code until required.
:param group: Name of the entry point group e.g. "datacube.plugins.io.read"
:returns: Dictionary String -> Driver Object
"""
def safe_load(ep):
from pkg_resources import DistributionNotFound
# pylint: disable=broad-except,bare-except
try:
driver_init = ep.load()
except DistributionNotFound:
# This happens when entry points were marked with extra features,
# but extra feature were not requested for installation
return None
except Exception as e:
_LOG.warning('Failed to resolve driver %s::%s', group, ep.name)
_LOG.warning('Error was: %s', repr(e))
return None
try:
driver = driver_init()
except Exception:
_LOG.warning('Exception during driver init, driver name: %s::%s', group, ep.name)
return None
if driver is None:
_LOG.warning('Driver init returned None, driver name: %s::%s', group, ep.name)
return driver
def resolve_all(group: str) -> Iterable[Tuple[str, Any]]:
from pkg_resources import iter_entry_points
for ep in iter_entry_points(group=group, name=None):
driver = safe_load(ep)
if driver is not None:
yield (ep.name, driver)
return dict((name, driver) for name, driver in resolve_all(group))