/usr/lib/python2.7/dist-packages/celery/backends/couchbase.py is in python-celery 3.1.20-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | # -*- coding: utf-8 -*-
"""
celery.backends.couchbase
~~~~~~~~~~~~~~~~~~~~~~~~~
CouchBase result store backend.
"""
from __future__ import absolute_import
import logging
try:
from couchbase import Couchbase
from couchbase.connection import Connection
from couchbase.exceptions import NotFoundError
except ImportError:
Couchbase = Connection = NotFoundError = None # noqa
from kombu.utils.url import _parse_url
from celery.exceptions import ImproperlyConfigured
from celery.utils.timeutils import maybe_timedelta
from .base import KeyValueStoreBackend
__all__ = ['CouchBaseBackend']
class CouchBaseBackend(KeyValueStoreBackend):
bucket = 'default'
host = 'localhost'
port = 8091
username = None
password = None
quiet = False
conncache = None
unlock_gil = True
timeout = 2.5
transcoder = None
# supports_autoexpire = False
def __init__(self, url=None, *args, **kwargs):
"""Initialize CouchBase backend instance.
:raises celery.exceptions.ImproperlyConfigured: if
module :mod:`couchbase` is not available.
"""
super(CouchBaseBackend, self).__init__(*args, **kwargs)
self.expires = kwargs.get('expires') or maybe_timedelta(
self.app.conf.CELERY_TASK_RESULT_EXPIRES)
if Couchbase is None:
raise ImproperlyConfigured(
'You need to install the couchbase library to use the '
'CouchBase backend.',
)
uhost = uport = uname = upass = ubucket = None
if url:
_, uhost, uport, uname, upass, ubucket, _ = _parse_url(url)
ubucket = ubucket.strip('/') if ubucket else None
config = self.app.conf.get('CELERY_COUCHBASE_BACKEND_SETTINGS', None)
if config is not None:
if not isinstance(config, dict):
raise ImproperlyConfigured(
'Couchbase backend settings should be grouped in a dict',
)
else:
config = {}
self.host = uhost or config.get('host', self.host)
self.port = int(uport or config.get('port', self.port))
self.bucket = ubucket or config.get('bucket', self.bucket)
self.username = uname or config.get('username', self.username)
self.password = upass or config.get('password', self.password)
self._connection = None
def _get_connection(self):
"""Connect to the Couchbase server."""
if self._connection is None:
kwargs = {'bucket': self.bucket, 'host': self.host}
if self.port:
kwargs.update({'port': self.port})
if self.username:
kwargs.update({'username': self.username})
if self.password:
kwargs.update({'password': self.password})
logging.debug('couchbase settings %r', kwargs)
self._connection = Connection(**kwargs)
return self._connection
@property
def connection(self):
return self._get_connection()
def get(self, key):
try:
return self.connection.get(key).value
except NotFoundError:
return None
def set(self, key, value):
self.connection.set(key, value)
def mget(self, keys):
return [self.get(key) for key in keys]
def delete(self, key):
self.connection.delete(key)
|