def _get_session()

in granule_ingester/granule_ingester/writers/CassandraStore.py [0:0]

• 18 lines of code
• 3 McCabe index (conditional complexity)

    def _get_session(self) -> Session:

        if self._username and self._password:
            auth_provider = PlainTextAuthProvider(username=self._username, password=self._password)
        else:
            auth_provider = None

        cluster = Cluster(contact_points=self._contact_points,
                          port=self._port,
                          # load_balancing_policy=
                          execution_profiles={
                              EXEC_PROFILE_DEFAULT: ExecutionProfile(
                                  request_timeout=60.0,
                                  retry_policy=RetryPolicy()
                              )
                          },
                          reconnection_policy=ConstantReconnectionPolicy(delay=5.0),
                          auth_provider=auth_provider)
        session = cluster.connect()
        session.set_keyspace(self._keyspace)
        return session