registry.py 3.4 KB

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