registry.py 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596
  1. from itertools import chain
  2. from django.utils.itercompat import is_iterable
  3. class Tags:
  4. """
  5. Built-in tags for internal checks.
  6. """
  7. admin = 'admin'
  8. async_support = 'async_support'
  9. caches = 'caches'
  10. compatibility = 'compatibility'
  11. database = 'database'
  12. models = 'models'
  13. security = 'security'
  14. signals = 'signals'
  15. sites = 'sites'
  16. staticfiles = 'staticfiles'
  17. templates = 'templates'
  18. translation = 'translation'
  19. urls = 'urls'
  20. class CheckRegistry:
  21. def __init__(self):
  22. self.registered_checks = set()
  23. self.deployment_checks = set()
  24. def register(self, check=None, *tags, **kwargs):
  25. """
  26. Can be used as a function or a decorator. Register given function
  27. `f` labeled with given `tags`. The function should receive **kwargs
  28. and return list of Errors and Warnings.
  29. Example::
  30. registry = CheckRegistry()
  31. @registry.register('mytag', 'anothertag')
  32. def my_check(apps, **kwargs):
  33. # ... perform checks and collect `errors` ...
  34. return errors
  35. # or
  36. registry.register(my_check, 'mytag', 'anothertag')
  37. """
  38. def inner(check):
  39. check.tags = tags
  40. checks = self.deployment_checks if kwargs.get('deploy') else self.registered_checks
  41. checks.add(check)
  42. return check
  43. if callable(check):
  44. return inner(check)
  45. else:
  46. if check:
  47. tags += (check,)
  48. return inner
  49. def run_checks(self, app_configs=None, tags=None, include_deployment_checks=False, databases=None):
  50. """
  51. Run all registered checks and return list of Errors and Warnings.
  52. """
  53. errors = []
  54. checks = self.get_checks(include_deployment_checks)
  55. if tags is not None:
  56. checks = [check for check in checks if not set(check.tags).isdisjoint(tags)]
  57. for check in checks:
  58. new_errors = check(app_configs=app_configs, databases=databases)
  59. assert is_iterable(new_errors), (
  60. "The function %r did not return a list. All functions registered "
  61. "with the checks registry must return a list." % check)
  62. errors.extend(new_errors)
  63. return errors
  64. def tag_exists(self, tag, include_deployment_checks=False):
  65. return tag in self.tags_available(include_deployment_checks)
  66. def tags_available(self, deployment_checks=False):
  67. return set(chain.from_iterable(
  68. check.tags for check in self.get_checks(deployment_checks)
  69. ))
  70. def get_checks(self, include_deployment_checks=False):
  71. checks = list(self.registered_checks)
  72. if include_deployment_checks:
  73. checks.extend(self.deployment_checks)
  74. return checks
  75. registry = CheckRegistry()
  76. register = registry.register
  77. run_checks = registry.run_checks
  78. tag_exists = registry.tag_exists