threading_checker.py 1.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. # Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
  2. # For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
  3. from astroid import nodes
  4. from pylint import interfaces
  5. from pylint.checkers import BaseChecker
  6. from pylint.checkers.utils import check_messages, safe_infer
  7. class ThreadingChecker(BaseChecker):
  8. """Checks for threading module
  9. - useless with lock - locking used in wrong way that has no effect (with threading.Lock():)
  10. """
  11. __implements__ = interfaces.IAstroidChecker
  12. name = "threading"
  13. LOCKS = frozenset(
  14. (
  15. "threading.Lock",
  16. "threading.RLock",
  17. "threading.Condition",
  18. "threading.Semaphore",
  19. "threading.BoundedSemaphore",
  20. )
  21. )
  22. msgs = {
  23. "W2101": (
  24. "'%s()' directly created in 'with' has no effect",
  25. "useless-with-lock",
  26. "Used when a new lock instance is created by using with statement "
  27. "which has no effect. Instead, an existing instance should be used to acquire lock.",
  28. ),
  29. }
  30. @check_messages("useless-with-lock")
  31. def visit_with(self, node: nodes.With) -> None:
  32. context_managers = (c for c, _ in node.items if isinstance(c, nodes.Call))
  33. for context_manager in context_managers:
  34. if isinstance(context_manager, nodes.Call):
  35. infered_function = safe_infer(context_manager.func)
  36. if infered_function is None:
  37. continue
  38. qname = infered_function.qname()
  39. if qname in self.LOCKS:
  40. self.add_message("useless-with-lock", node=node, args=qname)
  41. def register(linter):
  42. """required method to auto register this checker"""
  43. linter.register_checker(ThreadingChecker(linter))