valkeydb.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. # SPDX-License-Identifier: AGPL-3.0-or-later
  2. """Implementation of the valkey client (valkey-py_).
  3. .. _valkey-py: https://github.com/valkey-io/valkey-py
  4. This implementation uses the :ref:`settings valkey` setup from ``settings.yml``.
  5. A valkey DB connect can be tested by::
  6. >>> from searx import valkeydb
  7. >>> valkeydb.initialize()
  8. True
  9. >>> db = valkeydb.client()
  10. >>> db.set("foo", "bar")
  11. True
  12. >>> db.get("foo")
  13. b'bar'
  14. >>>
  15. """
  16. import os
  17. import pwd
  18. import logging
  19. import warnings
  20. import valkey
  21. from searx import get_setting
  22. _CLIENT = None
  23. logger = logging.getLogger(__name__)
  24. def client() -> valkey.Valkey:
  25. return _CLIENT
  26. def initialize():
  27. global _CLIENT # pylint: disable=global-statement
  28. if get_setting('redis.url'):
  29. warnings.warn("setting redis.url is deprecated, use valkey.url", DeprecationWarning)
  30. valkey_url = get_setting('valkey.url') or get_setting('redis.url')
  31. if not valkey_url:
  32. return False
  33. try:
  34. # create a client, but no connection is done
  35. _CLIENT = valkey.Valkey.from_url(valkey_url)
  36. # log the parameters as seen by the valkey lib, without the password
  37. kwargs = _CLIENT.get_connection_kwargs().copy()
  38. kwargs.pop('password', None)
  39. kwargs = ' '.join([f'{k}={v!r}' for k, v in kwargs.items()])
  40. logger.info("connecting to Valkey %s", kwargs)
  41. # check the connection
  42. _CLIENT.ping()
  43. # no error: the valkey connection is working
  44. logger.info("connected to Valkey")
  45. return True
  46. except valkey.exceptions.ValkeyError:
  47. _CLIENT = None
  48. _pw = pwd.getpwuid(os.getuid())
  49. logger.exception("[%s (%s)] can't connect valkey DB ...", _pw.pw_name, _pw.pw_uid)
  50. return False