celery__celery-9465

{
  "status": "ok",
  "filepath": "celery/backends/cassandra.py",
  "content": "\"\"\"Apache Cassandra result store backend using the DataStax driver.\"\"\"\nimport threading\n\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.utils.log import get_logger\n\nfrom .base import BaseBackend\n\ntry:  # pragma: no cover\n    import cassandra\n    import cassandra.auth\n    import cassandra.cluster\n    import cassandra.query\nexcept ImportError:\n    cassandra = None\n\n\n__all__ = ('CassandraBackend',)\n\nlogger = get_logger(__name__)\n\nE_NO_CASSANDRA = \"\"\"\nYou need to install the cassandra-driver library to\nuse the Cassandra backend.  See https://github.com/datastax/python-driver\n\"\"\"\n\nE_NO_SUCH_CASSANDRA_AUTH_PROVIDER = \"\"\"\nCASSANDRA_AUTH_PROVIDER you provided is not a valid auth_provider class.\nSee https://datastax.github.io/python-driver/api/cassandra/auth.html.\n\"\"\"\n\nE_CASSANDRA_MISCONFIGURED = 'Cassandra backend improperly configured.'\n\nE_CASSANDRA_NOT_CONFIGURED = 'Cassandra backend not configured.'\n\nQ_INSERT_RESULT = \"\"\"\nINSERT INTO {table} (\n    task_id, status, result, date_done, traceback, children) VALUES (\n        %s, %s, %s, %s, %s, %s) {expires};\n\"\"\"\n\nQ_SELECT_RESULT = \"\"\"\nSELECT status, result, date_done, traceback, children\nFROM {table}\nWHERE task_id=%s\nLIMIT 1\n\"\"\"\n\nQ_CREATE_RESULT_TABLE = \"\"\"\nCREATE TABLE {table} (\n    task_id text,\n    status text,\n    result blob,\n    date_done timestamp,\n    traceback blob,\n    children blob,\n    PRIMARY KEY ((task_id), date_done)\n) WITH CLUSTERING ORDER BY (date_done DESC);\n\"\"\"\n\nQ_EXPIRES = \"\"\"\n    USING TTL {0}\n\"\"\"\n\n\ndef buf_t(x):\n    return bytes(x, 'utf8')\n\n\nclass CassandraBackend(BaseBackend):\n    \"\"\"Cassandra/AstraDB backend utilizing DataStax driver.\n\n    Raises:\n        celery.exceptions.ImproperlyConfigured:\n            if module :pypi:`cassandra-driver` is not available,\n            or not-exactly-one of the :setting:`cassandra_servers` and\n            the :setting:`cassandra_secure_bundle_path` settings is set.\n    \"\"\"\n\n    #: List of Cassandra servers with format: ``hostname``.\n    servers = None\n    #: Location of the secure connect bundle zipfile (absolute path).\n    bundle_path = None\n\n    supports_autoexpire = True      # autoexpire supported via entry_ttl\n\n    def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,\n                 port=9042, bundle_path=None, **kwargs):\n        super().__init__(**kwargs)\n\n        if not cassandra:\n            raise ImproperlyConfigured(E_NO_CASSANDRA)\n\n        conf = self.app.conf\n        self.servers = servers or conf.get('cassandra_servers', None)\n        self.bundle_path = bundle_path or conf.get(\n            'cassandra_secure_bundle_path', None)\n        self.port = port or conf.get('cassandra_port', None)\n        self.keyspace = keyspace or conf.get('cassandra_keyspace', None)\n        self.table = table or conf.get('cassandra_table', None)\n        self.cassandra_options = conf.get('cassandra_options', {})\n\n        # either servers or bundle path must be provided...\n        db_directions = self.servers or self.bundle_path\n        if not db_directions or not self.keyspace or not self.table:\n            raise ImproperlyConfigured(E_CASSANDRA_NOT_CONFIGURED)\n        # ...but not both:\n        if self.servers and self.bundle_path:\n            raise ImproperlyConfigured(E_CASSANDRA_MISCONFIGURED)\n\n        expires = entry_ttl or conf.get('cassandra_entry_ttl', None)\n\n        self.cqlexpires = (\n            Q_EXPIRES.format(expires) if expires is not None else '')\n\n        read_cons = conf.get('cassandra_read_consistency') or 'LOCAL_QUORUM'\n        write_cons = conf.get('cassandra_write_consistency') or 'LOCAL_QUORUM'\n\n        self.read_consistency = getattr(\n            cassandra.ConsistencyLevel, read_cons,\n            cassandra.ConsistencyLevel.LOCAL_QUORUM)\n        self.write_consistency = getattr(\n            cassandra.ConsistencyLevel, write_cons,\n            cassandra.ConsistencyLevel.LOCAL_QUORUM)\n\n        self.auth_provider = None\n        auth_provider = conf.get('cassandra_auth_provider', None)\n        auth_kwargs = conf.get('cassandra_auth_kwargs', None)\n        if auth_provider and auth_kwargs:\n            auth_provider_class = getattr(cassandra.auth, auth_provider, None)\n            if not auth_provider_class:\n                raise ImproperlyConfigured(E_NO_SUCH_CASSANDRA_AUTH_PROVIDER)\n            self.auth_provider = auth_provider_class(**auth_kwargs)\n\n        self._cluster = None\n        self._session = None\n        self._write_stmt = None\n        self._read_stmt = None\n        self._lock = threading.RLock()\n\n    def _get_connection(self, write=False):\n        \"\"\"Prepare the connection for action.\n\n        Arguments:\n            write (bool): are we a writer?\n        \"\"\"\n        if self._session is not None:\n            return\n        self._lock.acquire()",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 256,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "celery/backends/cassandra.py",
  "content": "        try:\n            if self._session is not None:\n                return\n            # using either 'servers' or 'bundle_path' here:\n            if self.servers:\n                self._cluster = cassandra.cluster.Cluster(\n                    self.servers, port=self.port,\n                    auth_provider=self.auth_provider,\n                    **self.cassandra_options)\n            else:\n                # 'bundle_path' is guaranteed to be set\n                self._cluster = cassandra.cluster.Cluster(\n                    cloud={\n                        'secure_connect_bundle': self.bundle_path,\n                    },\n                    auth_provider=self.auth_provider,\n                    **self.cassandra_options)\n            self._session = self._cluster.connect(self.keyspace)\n\n            # We're forced to do concatenation below, as formatting would\n            # blow up on superficial %s that'll be processed by Cassandra\n            self._write_stmt = cassandra.query.SimpleStatement(\n                Q_INSERT_RESULT.format(\n                    table=self.table, expires=self.cqlexpires),\n            )\n            self._write_stmt.consistency_level = self.write_consistency\n\n            self._read_stmt = cassandra.query.SimpleStatement(\n                Q_SELECT_RESULT.format(table=self.table),\n            )\n            self._read_stmt.consistency_level = self.read_consistency\n\n            if write:\n                # Only possible writers \"workers\" are allowed to issue\n                # CREATE TABLE.  This is to prevent conflicting situations\n                # where both task-creator and task-executor would issue it\n                # at the same time.\n\n                # Anyway; if you're doing anything critical, you should\n                # have created this table in advance, in which case\n                # this query will be a no-op (AlreadyExists)\n                make_stmt = cassandra.query.SimpleStatement(\n                    Q_CREATE_RESULT_TABLE.format(table=self.table),\n                )\n                make_stmt.consistency_level = self.write_consistency\n\n                try:\n                    self._session.execute(make_stmt)\n                except cassandra.AlreadyExists:\n                    pass\n\n        except cassandra.OperationTimedOut:\n            # a heavily loaded or gone Cassandra cluster failed to respond.\n            # leave this class in a consistent state\n            if self._cluster is not None:\n                self._cluster.shutdown()     # also shuts down _session\n\n            self._cluster = None\n            self._session = None\n            raise   # we did fail after all - reraise\n        finally:\n            self._lock.release()\n\n    def _store_result(self, task_id, result, state,\n                      traceback=None, request=None, **kwargs):\n        \"\"\"Store return value and state of an executed task.\"\"\"\n        self._get_connection(write=True)\n\n        self._session.execute(self._write_stmt, (\n            task_id,\n            state,\n            buf_t(self.encode(result)),\n            self.app.now(),\n            buf_t(self.encode(traceback)),\n            buf_t(self.encode(self.current_task_children(request)))\n        ))\n\n    def as_uri(self, include_password=True):\n        return 'cassandra://'\n\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task meta-data for a task by id.\"\"\"\n        self._get_connection()\n\n        res = self._session.execute(self._read_stmt, (task_id, )).one()\n        if not res:\n            return {'status': states.PENDING, 'result': None}\n\n        status, result, date_done, traceback, children = res\n\n        return self.meta_from_decoded({\n            'task_id': task_id,\n            'status': status,\n            'result': self.decode(result),\n            'date_done': date_done,\n            'traceback': self.decode(traceback),\n            'children': self.decode(children),\n        })\n\n    def __reduce__(self, args=(), kwargs=None):\n        kwargs = {} if not kwargs else kwargs\n        kwargs.update(\n            {'servers': self.servers,\n             'keyspace': self.keyspace,\n             'table': self.table})\n        return super().__reduce__(args, kwargs)",
  "start_line": 151,
  "end_line": 256,
  "total_lines": 256,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "EE\n======================================================================\nERROR: test_port_from_config (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 24, in test_port_from_config\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 116, in __init__\n    self.serializer = serializer or conf.result_serializer\nAttributeError: 'dict' object has no attribute 'result_serializer'\n\n======================================================================\nERROR: test_port_from_init (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 41, in test_port_from_init\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 116, in __init__\n    self.serializer = serializer or conf.result_serializer\nAttributeError: 'dict' object has no attribute 'result_serializer'\n\n----------------------------------------------------------------------\nRan 2 tests in 0.001s\n\nFAILED (errors=2)\n",
  "details": {
    "stdout": "",
    "stderr": "EE\n======================================================================\nERROR: test_port_from_config (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 24, in test_port_from_config\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 116, in __init__\n    self.serializer = serializer or conf.result_serializer\nAttributeError: 'dict' object has no attribute 'result_serializer'\n\n======================================================================\nERROR: test_port_from_init (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 41, in test_port_from_init\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 116, in __init__\n    self.serializer = serializer or conf.result_serializer\nAttributeError: 'dict' object has no attribute 'result_serializer'\n\n----------------------------------------------------------------------\nRan 2 tests in 0.001s\n\nFAILED (errors=2)\n",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "EE\n======================================================================\nERROR: test_port_from_config (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 31, in test_port_from_config\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 120, in __init__\n    cmax = max_cached_results or conf.result_cache_max\nAttributeError: 'Config' object has no attribute 'result_cache_max'\n\n======================================================================\nERROR: test_port_from_init (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 47, in test_port_from_init\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 120, in __init__\n    cmax = max_cached_results or conf.result_cache_max\nAttributeError: 'Config' object has no attribute 'result_cache_max'\n\n----------------------------------------------------------------------\nRan 2 tests in 0.001s\n\nFAILED (errors=2)\n",
  "details": {
    "stdout": "",
    "stderr": "EE\n======================================================================\nERROR: test_port_from_config (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 31, in test_port_from_config\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 120, in __init__\n    cmax = max_cached_results or conf.result_cache_max\nAttributeError: 'Config' object has no attribute 'result_cache_max'\n\n======================================================================\nERROR: test_port_from_init (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 47, in test_port_from_init\n  File \"/workspace/celery/backends/cassandra.py\", line 90, in __init__\n    super().__init__(**kwargs)\n  File \"/workspace/celery/backends/base.py\", line 120, in __init__\n    cmax = max_cached_results or conf.result_cache_max\nAttributeError: 'Config' object has no attribute 'result_cache_max'\n\n----------------------------------------------------------------------\nRan 2 tests in 0.001s\n\nFAILED (errors=2)\n",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "F.\n======================================================================\nFAIL: test_port_from_config (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 36, in test_port_from_config\nAssertionError: 9042 != 61003\n\n----------------------------------------------------------------------\nRan 2 tests in 0.003s\n\nFAILED (failures=1)\n",
  "details": {
    "stdout": "",
    "stderr": "F.\n======================================================================\nFAIL: test_port_from_config (__main__.TestCassandraPort)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"<stdin>\", line 36, in test_port_from_config\nAssertionError: 9042 != 61003\n\n----------------------------------------------------------------------\nRan 2 tests in 0.003s\n\nFAILED (failures=1)\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "",
  "stderr": ".\n----------------------------------------------------------------------\nRan 1 test in 0.001s\n\nOK\n",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "",
  "stderr": ".\n----------------------------------------------------------------------\nRan 1 test in 0.001s\n\nOK\n",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "celery/backends/cassandra.py",
  "occurrences": 1,
  "strategy": "exact",
  "diff": "--- a/celery/backends/cassandra.py\n+++ b/celery/backends/cassandra.py\n@@ -86,7 +86,7 @@\n     supports_autoexpire = True      # autoexpire supported via entry_ttl\n \n     def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,\n-                 port=9042, bundle_path=None, **kwargs):\n+                 port=None, bundle_path=None, **kwargs):\n         super().__init__(**kwargs)\n \n         if not cassandra:\n@@ -97,6 +97,8 @@\n         self.bundle_path = bundle_path or conf.get(\n             'cassandra_secure_bundle_path', None)\n         self.port = port or conf.get('cassandra_port', None)\n+        if self.port is not None:\n+            self.port = int(self.port)\n         self.keyspace = keyspace or conf.get('cassandra_keyspace', None)\n         self.table = table or conf.get('cassandra_table', None)\n         self.cassandra_options = conf.get('cassandra_options', {})\n",
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "python3: can't open file '/tmp/repro.py': [Errno 2] No such file or directory\n",
  "details": {
    "stdout": "",
    "stderr": "python3: can't open file '/tmp/repro.py': [Errno 2] No such file or directory\n",
    "exit_code": 2
  }
}

{
  "status": "ok",
  "stdout": "",
  "stderr": "...\n----------------------------------------------------------------------\nRan 3 tests in 0.003s\n\nOK\n",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "ERROR: file or directory not found: tests/unit/backends/test_cassandra.py\n\n",
  "details": {
    "stdout": "\nno tests ran in 0.00s\n",
    "stderr": "ERROR: file or directory not found: tests/unit/backends/test_cassandra.py\n\n",
    "exit_code": 4
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "find: ‘tests’: No such file or directory\n",
  "details": {
    "stdout": "",
    "stderr": "find: ‘tests’: No such file or directory\n",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "ls: cannot access 'tests': No such file or directory\n",
  "details": {
    "stdout": "",
    "stderr": "ls: cannot access 'tests': No such file or directory\n",
    "exit_code": 2
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "ls: cannot access 'test': No such file or directory\nls: cannot access 'tests': No such file or directory\n",
  "details": {
    "stdout": "",
    "stderr": "ls: cannot access 'test': No such file or directory\nls: cannot access 'tests': No such file or directory\n",
    "exit_code": 2
  }
}

{
  "status": "ok",
  "stdout": "CONTRIBUTING.rst\nCONTRIBUTORS.txt\nChangelog.rst\nLICENSE\nMANIFEST.in\nMakefile\nREADME.rst\nSECURITY.md\nTODO\nbandit.json\ncelery/\ndocker/\ndocs/\nexamples/\nextra/\nhelm-chart/\npyproject.toml\nrequirements/\nsetup.cfg\nsetup.py*\nt/\ntox.ini\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "./t/unit/backends/test_cassandra.py:19:class test_CassandraBackend:\n./t/unit/backends/test_cassandra.py:36:                mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:52:        mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:54:        mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:60:            mod.CassandraBackend(\n./t/unit/backends/test_cassandra.py:69:            mod.CassandraBackend(\n./t/unit/backends/test_cassandra.py:96:        x = mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:103:        from celery.backends.cassandra import CassandraBackend\n./t/unit/backends/test_cassandra.py:104:        assert loads(dumps(CassandraBackend(app=self.app)))\n./t/unit/backends/test_cassandra.py:111:        x = mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:133:        x = mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:142:        x = mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:170:        x = mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:205:        x = mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:225:        x = mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:250:        mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:258:            mod.CassandraBackend(app=self.app)\n./t/unit/backends/test_cassandra.py:270:        mod.CassandraBackend(app=self.app)\n./celery/app/backends.py:25:    'cassandra': 'celery.backends.cassandra:CassandraBackend',\n./celery/backends/cassandra.py:19:__all__ = ('CassandraBackend',)\n./celery/backends/cassandra.py:71:class CassandraBackend(BaseBackend):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "............                                                             [100%]\n12 passed in 0.04s\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 1, in <module>\nModuleNotFoundError: No module named 'cassandra'\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 1, in <module>\nModuleNotFoundError: No module named 'cassandra'\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": " M celery/backends/cassandra.py\ndiff --git a/celery/backends/cassandra.py b/celery/backends/cassandra.py\nindex 0eb37f3..e2e7829 100644\n--- a/celery/backends/cassandra.py\n+++ b/celery/backends/cassandra.py\n@@ -86,7 +86,7 @@ class CassandraBackend(BaseBackend):\n     supports_autoexpire = True      # autoexpire supported via entry_ttl\n \n     def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,\n-                 port=9042, bundle_path=None, **kwargs):\n+                 port=None, bundle_path=None, **kwargs):\n         super().__init__(**kwargs)\n \n         if not cassandra:\n@@ -97,6 +97,8 @@ class CassandraBackend(BaseBackend):\n         self.bundle_path = bundle_path or conf.get(\n             'cassandra_secure_bundle_path', None)\n         self.port = port or conf.get('cassandra_port', None)\n+        if self.port is not None:\n+            self.port = int(self.port)\n         self.keyspace = keyspace or conf.get('cassandra_keyspace', None)\n         self.table = table or conf.get('cassandra_table', None)\n         self.cassandra_options = conf.get('cassandra_options', {})\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "patch_size": 1069,
  "files_changed": 1
}