1from plain.utils.inspect import func_accepts_kwargs
2from plain.utils.itercompat import is_iterable
3
4
5class CheckRegistry:
6 def __init__(self):
7 self.registered_checks = set()
8 self.deployment_checks = set()
9
10 def register(self, check=None, deploy=False):
11 """
12 Can be used as a function or a decorator. Register given function
13 `f`. The function should receive **kwargs
14 and return list of Errors and Warnings.
15
16 Example::
17
18 registry = CheckRegistry()
19 @registry.register('mytag', 'anothertag')
20 def my_check(package_configs, **kwargs):
21 # ... perform checks and collect `errors` ...
22 return errors
23 # or
24 registry.register(my_check, 'mytag', 'anothertag')
25 """
26
27 def inner(check):
28 if not func_accepts_kwargs(check):
29 raise TypeError(
30 "Check functions must accept keyword arguments (**kwargs)."
31 )
32 checks = self.deployment_checks if deploy else self.registered_checks
33 checks.add(check)
34 return check
35
36 if callable(check):
37 return inner(check)
38 else:
39 return inner
40
41 def run_checks(
42 self,
43 package_configs=None,
44 include_deployment_checks=False,
45 databases=None,
46 ):
47 """
48 Run all registered checks and return list of Errors and Warnings.
49 """
50 errors = []
51 checks = self.get_checks(include_deployment_checks)
52
53 for check in checks:
54 new_errors = check(package_configs=package_configs, databases=databases)
55 if not is_iterable(new_errors):
56 raise TypeError(
57 f"The function {check!r} did not return a list. All functions "
58 "registered with the checks registry must return a list.",
59 )
60 errors.extend(new_errors)
61 return errors
62
63 def get_checks(self, include_deployment_checks=False):
64 checks = list(self.registered_checks)
65 if include_deployment_checks:
66 checks.extend(self.deployment_checks)
67 return checks
68
69
70registry = CheckRegistry()
71register = registry.register
72run_checks = registry.run_checks