title: Flask Cassandra Connection with Celery
body: <p>I am trying to create cassandra connection when the celery worker process starts and close the connections when the worker shuts down.</p>
  
  <p>I am using the below code for it.</p>
  
  <pre><code>thread_local = threading.local()
  
  @worker_process_init.connect
  def open_cassandra_session(*args, **kwargs):
    auth_provider = DSEPlainTextAuthProvider(username='simpl1', password='********')
    cluster = Cluster( ['public_ip1','public_ip2','public_ip3'], port=9042, auth_provider=auth_provider, address_translator=IdentityTranslator())
    session = cluster.connect('keyspace1')
    thread_local.cassandra_session = session
  
  @worker_process_shutdown.connect
  def close_cassandra_session(*args,**kwargs):
    session = thread_local.cassandra_session
    session.shutdown()
    thread_local.cassandra_session = None
  </code></pre>
  
  <p>When I start the celery worker, I get the below error</p>
  
  <pre><code>[2017-06-28 14:39:31,805: WARNING/PoolWorker-1] Downgrading core protocol version from 65 to 4 for 54.255.180.2. To avoid this, it is best practice to explicitly set Cluster(protocol_version) to the version supported by your cluster. http://docs.datastax.com/en/developer/python-driver-dse/latest/api/dse/cluster.html#dse.cluster.Cluster.protocol_version
  [2017-06-28 14:39:32,268: INFO/PoolWorker-1] Using datacenter 'ap-southeast_1_cassandra' for DCAwareRoundRobinPolicy (via host '54.255.180.2'); if incorrect, please specify a local_dc to the constructor, or limit contact points to local cluster nodes
  [2017-06-28 14:39:32,268: INFO/PoolWorker-1] New Cassandra host &lt;Host: 172.31.0.4 ap-southeast_1_cassandra&gt; discovered
  [2017-06-28 14:39:32,270: INFO/PoolWorker-1] New Cassandra host &lt;Host: 172.31.10.93 ap-southeast_1_cassandra&gt; discovered
  [2017-06-28 14:39:32,270: INFO/PoolWorker-1] New Cassandra host &lt;Host: 172.31.20.239 ap-southeast_1_cassandra&gt; discovered
  [2017-06-28 14:39:32,270: INFO/PoolWorker-1] Cassandra host 54.169.86.45 removed
  [2017-06-28 14:39:32,271: INFO/PoolWorker-1] Cassandra host 54.169.132.4 removed
  [2017-06-28 14:39:32,271: INFO/PoolWorker-1] Cassandra host 54.169.10.56 removed
  [2017-06-28 14:39:32,659: INFO/MainProcess] mingle: all alone
  [2017-06-28 14:39:32,672: INFO/MainProcess] celery@SanthavathisMBP ready.
  [2017-06-28 14:39:37,578: WARNING/PoolWorker-1] Failed to create connection pool for new host 172.31.0.4:
  Traceback (most recent call last):
    File "dse/cluster.py", line 2217, in dse.cluster.Session.add_or_renew_pool.run_add_or_renew_pool (dse/cluster.c:41176)
      new_pool = HostConnection(host, distance, self)
    File "dse/hosts.py", line 329, in dse.hosts.HostConnection.__init__ (dse/hosts.c:5903)
      self._connection = session.cluster.connection_factory(host.address)
    File "dse/cluster.py", line 1025, in dse.cluster.Cluster.connection_factory (dse/cluster.c:14184)
      return self.connection_class.factory(address, self.connect_timeout, *args, **kwargs)
    File "dse/connection.py", line 382, in dse.connection.Connection.factory (dse/connection.c:8213)
      conn = cls(host, *args, **kwargs)
    File "/Users/santha/Downloads/flask/lib/python2.7/site-packages/dse/io/asyncorereactor.py", line 294, in __init__
      self._connect_socket()
    File "dse/connection.py", line 421, in dse.connection.Connection._connect_socket (dse/connection.c:9522)
      raise socket.error(sockerr.errno, "Tried connecting to %s. Last error: %s" % ([a[4] for a in addresses], sockerr.strerror or sockerr))
  error: [Errno None] Tried connecting to [('172.31.0.4', 9042)]. Last error: timed out
  [2017-06-28 14:39:37,898: WARNING/PoolWorker-1] Failed to create connection pool for new host 172.31.20.239:
  Traceback (most recent call last):
    File "dse/cluster.py", line 2217, in dse.cluster.Session.add_or_renew_pool.run_add_or_renew_pool (dse/cluster.c:41176)
      new_pool = HostConnection(host, distance, self)
    File "dse/hosts.py", line 329, in dse.hosts.HostConnection.__init__ (dse/hosts.c:5903)
      self._connection = session.cluster.connection_factory(host.address)
    File "dse/cluster.py", line 1025, in dse.cluster.Cluster.connection_factory (dse/cluster.c:14184)
      return self.connection_class.factory(address, self.connect_timeout, *args, **kwargs)
    File "dse/connection.py", line 382, in dse.connection.Connection.factory (dse/connection.c:8213)
      conn = cls(host, *args, **kwargs)
    File "/Users/santha/Downloads/flask/lib/python2.7/site-packages/dse/io/asyncorereactor.py", line 294, in __init__
      self._connect_socket()
    File "dse/connection.py", line 421, in dse.connection.Connection._connect_socket (dse/connection.c:9522)
      raise socket.error(sockerr.errno, "Tried connecting to %s. Last error: %s" % ([a[4] for a in addresses], sockerr.strerror or sockerr))
  error: [Errno None] Tried connecting to [('172.31.20.239', 9042)]. Last error: timed out
  [2017-06-28 14:39:37,900: WARNING/PoolWorker-1] Host 172.31.0.4 has been marked down
  [2017-06-28 14:39:37,901: WARNING/PoolWorker-1] Host 172.31.20.239 has been marked down
  [2017-06-28 14:39:42,588: WARNING/PoolWorker-1] Failed to create connection pool for new host 172.31.10.93:
  Traceback (most recent call last):
    File "dse/cluster.py", line 2217, in dse.cluster.Session.add_or_renew_pool.run_add_or_renew_pool (dse/cluster.c:41176)
      new_pool = HostConnection(host, distance, self)
    File "dse/hosts.py", line 329, in dse.hosts.HostConnection.__init__ (dse/hosts.c:5903)
      self._connection = session.cluster.connection_factory(host.address)
    File "dse/cluster.py", line 1025, in dse.cluster.Cluster.connection_factory (dse/cluster.c:14184)
      return self.connection_class.factory(address, self.connect_timeout, *args, **kwargs)
    File "dse/connection.py", line 382, in dse.connection.Connection.factory (dse/connection.c:8213)
      conn = cls(host, *args, **kwargs)
    File "/Users/santha/Downloads/flask/lib/python2.7/site-packages/dse/io/asyncorereactor.py", line 294, in __init__
      self._connect_socket()
    File "dse/connection.py", line 421, in dse.connection.Connection._connect_socket (dse/connection.c:9522)
      raise socket.error(sockerr.errno, "Tried connecting to %s. Last error: %s" % ([a[4] for a in addresses], sockerr.strerror or sockerr))
  error: [Errno None] Tried connecting to [('172.31.10.93', 9042)]. Last error: timed out
  [2017-06-28 14:39:43,933: WARNING/PoolWorker-1] Error attempting to reconnect to 172.31.0.4, scheduling retry in 2.0 seconds: [Errno None] Tried connecting to [('172.31.0.4', 9042)]. Last error: timed out
  [2017-06-28 14:39:43,934: WARNING/PoolWorker-1] Host 172.31.10.93 has been marked down
  </code></pre>
  
  <p>It is disregarding the address_translator=IdentityTranslator() setting which means <strong>do no address translation</strong> and <strong>use the addresses provided in the cluster configuration</strong>. Since my cassandra cluster is in AWS, and flask server runs outside of AWS, I need to use the public IP to connect to the cassandra nodes, but it is trying to connect to the Private IP.</p>