diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index a9c1d00e97..db12325d36 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -4283,75 +4283,35 @@ class ResponseFuture(object):
        :meth:`.add_callbacks()`.
     """
 
-    query = None
-    """
-    The :class:`~.Statement` instance that is being executed through this
-    :class:`.ResponseFuture`.
-    """
-
-    is_schema_agreed = True
-    """
-    For DDL requests, this may be set ``False`` if the schema agreement poll after the response fails.
-
-    Always ``True`` for non-DDL requests.
-    """
-
-    request_encoded_size = None
-    """
-    Size of the request message sent
-    """
-
-    coordinator_host = None
-    """
-    The host from which we received a response
-    """
-
-    attempted_hosts = None
-    """
-    A list of hosts tried, including all speculative executions, retries, and pages
-    """
-
-    session = None
-    row_factory = None
-    message = None
-    default_timeout = None
-
-    _retry_policy = None
-    _profile_manager = None
-
-    _req_id = None
-    _final_result = _NOT_SET
-    _col_names = None
-    _col_types = None
-    _final_exception = None
-    _query_traces = None
-    _callbacks = None
-    _errbacks = None
-    _current_host = None
-    _connection = None
-    _query_retries = 0
-    _start_time = None
-    _metrics = None
-    _paging_state = None
-    _custom_payload = None
-    _warnings = None
-    _timer = None
-    _protocol_handler = ProtocolHandler
-    _spec_execution_plan = NoSpeculativeExecutionPlan()
-    _continuous_paging_session = None
-    _host = None
-
-    _warned_timeout = False
+    __slots__ = (
+        # Public attributes
+        'query', 'is_schema_agreed', 'request_encoded_size', 'coordinator_host',
+        'attempted_hosts', 'session', 'row_factory', 'message', 'timeout',
+        'prepared_statement', 'query_plan',
+        # Private attributes
+        '_retry_policy', '_req_id', '_final_result', '_col_names', '_col_types',
+        '_final_exception', '_query_traces', '_callbacks', '_errbacks',
+        '_current_host', '_connection', '_query_retries', '_start_time',
+        '_metrics', '_paging_state', '_custom_payload', '_warnings', '_timer',
+        '_protocol_handler', '_spec_execution_plan', '_continuous_paging_session',
+        '_host', '_load_balancer', '_callback_lock', '_event', '_errors',
+        '_continuous_paging_state'
+    )
 
     def __init__(self, session, message, query, timeout, metrics=None, prepared_statement=None,
                  retry_policy=RetryPolicy(), row_factory=None, load_balancer=None, start_time=None,
                  speculative_execution_plan=None, continuous_paging_state=None, host=None):
+        # Initialize attributes with default values
+        self.query = query
+        self.is_schema_agreed = True
+        self.request_encoded_size = None
+        self.coordinator_host = None
+        self.attempted_hosts = []
         self.session = session
         # TODO: normalize handling of retry policy and row factory
         self.row_factory = row_factory or session.row_factory
         self._load_balancer = load_balancer or session.cluster._default_load_balancing_policy
         self.message = message
-        self.query = query
         self.timeout = timeout
         self._retry_policy = retry_policy
         self._metrics = metrics
@@ -4359,13 +4319,30 @@ def __init__(self, session, message, query, timeout, metrics=None, prepared_stat
         self._callback_lock = Lock()
         self._start_time = start_time or time.time()
         self._host = host
-        self._spec_execution_plan = speculative_execution_plan or self._spec_execution_plan
+        self._spec_execution_plan = speculative_execution_plan or NoSpeculativeExecutionPlan()
+        self._protocol_handler = ProtocolHandler
+
+        # Initialize other private attributes
+        self._req_id = None
+        self._final_result = _NOT_SET
+        self._col_names = None
+        self._col_types = None
+        self._final_exception = None
+        self._query_traces = None
+        self._current_host = None
+        self._connection = None
+        self._query_retries = 0
+        self._paging_state = None
+        self._custom_payload = None
+        self._warnings = None
+        self._timer = None
+        self._continuous_paging_session = None
+
         self._make_query_plan()
         self._event = Event()
         self._errors = {}
         self._callbacks = []
         self._errbacks = []
-        self.attempted_hosts = []
         self._start_timer()
         self._continuous_paging_state = continuous_paging_state
 
diff --git a/cassandra/connection.py b/cassandra/connection.py
index 9ac02c9776..eb82cfdfbc 100644
--- a/cassandra/connection.py
+++ b/cassandra/connection.py
@@ -370,6 +370,8 @@ def __repr__(self):
 
 
 class _Frame(object):
+    __slots__ = ('version', 'flags', 'stream', 'opcode', 'body_offset', 'end_pos')
+
     def __init__(self, version, flags, stream, opcode, body_offset, end_pos):
         self.version = version
         self.flags = flags
diff --git a/cassandra/metadata.py b/cassandra/metadata.py
index b85308449e..4aa1060bdc 100644
--- a/cassandra/metadata.py
+++ b/cassandra/metadata.py
@@ -1647,31 +1647,18 @@ def escape_name(name):
 class ColumnMetadata(object):
     """
     A representation of a single column in a table.
-    """
-
-    table = None
-    """ The :class:`.TableMetadata` this column belongs to. """
-
-    name = None
-    """ The string name of this column. """
-
-    cql_type = None
-    """
-    The CQL type for the column.
-    """
-
-    is_static = False
-    """
-    If this column is static (available in Cassandra 2.1+), this will
-    be :const:`True`, otherwise :const:`False`.
-    """
 
-    is_reversed = False
-    """
-    If this column is reversed (DESC) as in clustering order
+    Attributes:
+        table: The :class:`.TableMetadata` this column belongs to.
+        name: The string name of this column.
+        cql_type: The CQL type for the column.
+        is_static: If this column is static (available in Cassandra 2.1+), this
+    will be :const:`True`, otherwise :const:`False`.
+        is_reversed: If this column is reversed (DESC) as in clustering order.
+        _cass_type: Internal cache for the cassandra type.
     """
 
-    _cass_type = None
+    __slots__ = ('table', 'name', 'cql_type', 'is_static', 'is_reversed', '_cass_type')
 
     def __init__(self, table_metadata, column_name, cql_type, is_static=False, is_reversed=False):
         self.table = table_metadata
@@ -1679,6 +1666,7 @@ def __init__(self, table_metadata, column_name, cql_type, is_static=False, is_re
         self.cql_type = cql_type
         self.is_static = is_static
         self.is_reversed = is_reversed
+        self._cass_type = None
 
     def __str__(self):
         return "%s %s" % (self.name, self.cql_type)
@@ -1687,21 +1675,16 @@ def __str__(self):
 class IndexMetadata(object):
     """
     A representation of a secondary index on a column.
-    """
-    keyspace_name = None
-    """ A string name of the keyspace. """
 
-    table_name = None
-    """ A string name of the table this index is on. """
-
-    name = None
-    """ A string name for the index. """
-
-    kind = None
-    """ A string representing the kind of index (COMPOSITE, CUSTOM,...). """
+    Attributes:
+        keyspace_name: A string name of the keyspace.
+        table_name: A string name of the table this index is on.
+        name: A string name for the index.
+        kind: A string representing the kind of index (COMPOSITE, CUSTOM, ...).
+        index_options: A dict of index options.
+    """
 
-    index_options = {}
-    """ A dict of index options. """
+    __slots__ = ('keyspace_name', 'table_name', 'name', 'kind', 'index_options')
 
     def __init__(self, keyspace_name, table_name, index_name, kind, index_options):
         self.keyspace_name = keyspace_name
diff --git a/cassandra/protocol.py b/cassandra/protocol.py
index e574965de8..f4ee9d16d3 100644
--- a/cassandra/protocol.py
+++ b/cassandra/protocol.py
@@ -85,10 +85,12 @@ def __init__(cls, name, bases, dct):
 
 
 class _MessageType(object, metaclass=_RegisterMessageType):
+    __slots__ = ('custom_payload', 'tracing', 'allow_beta_protocol_version')
 
-    tracing = False
-    custom_payload = None
-    warnings = None
+    def __init__(self):
+        self.custom_payload = None
+        self.tracing = False
+        self.allow_beta_protocol_version = False
 
     def update_custom_payload(self, other):
         if other:
@@ -102,18 +104,28 @@ def __repr__(self):
         return '<%s(%s)>' % (self.__class__.__name__, ', '.join('%s=%r' % i for i in _get_params(self)))
 
 
+class _DecodableMessageType(_MessageType):
+    """Base class for messages that can be decoded and receive protocol attributes"""
+    __slots__ = ('stream_id', 'trace_id', 'warnings')
+
+
 def _get_params(message_obj):
     base_attrs = dir(_MessageType)
+    # Use __slots__ to get attributes since we no longer have __dict__
+    all_slots = []
+    for cls in type(message_obj).__mro__:
+        if hasattr(cls, '__slots__'):
+            all_slots.extend(cls.__slots__)
     return (
-        (n, a) for n, a in message_obj.__dict__.items()
-        if n not in base_attrs and not n.startswith('_') and not callable(a)
+        (n, getattr(message_obj, n, None)) for n in all_slots
+        if n not in base_attrs and not n.startswith('_') and hasattr(message_obj, n)
     )
 
 
 error_classes = {}
 
 
-class ErrorMessage(_MessageType, Exception):
+class ErrorMessage(Exception):
     opcode = 0x00
     name = 'ERROR'
     summary = 'Unknown'
@@ -407,6 +419,10 @@ class ClientWriteError(RequestExecutionException):
     error_code = 0x8000
 
 
+# Manually register ErrorMessage since it doesn't use _RegisterMessageType metaclass
+register_class(ErrorMessage)
+
+
 class StartupMessage(_MessageType):
     opcode = 0x01
     name = 'STARTUP'
@@ -418,6 +434,7 @@ class StartupMessage(_MessageType):
     ))
 
     def __init__(self, cqlversion, options):
+        super().__init__()
         self.cqlversion = cqlversion
         self.options = options
 
@@ -427,7 +444,9 @@ def send_body(self, f, protocol_version):
         write_stringmap(f, optmap)
 
 
-class ReadyMessage(_MessageType):
+class ReadyMessage(_DecodableMessageType):
+    __slots__ = ()
+
     opcode = 0x02
     name = 'READY'
 
@@ -436,11 +455,14 @@ def recv_body(cls, *args):
         return cls()
 
 
-class AuthenticateMessage(_MessageType):
+class AuthenticateMessage(_DecodableMessageType):
+    __slots__ = ('authenticator',)
+
     opcode = 0x03
     name = 'AUTHENTICATE'
 
     def __init__(self, authenticator):
+        super().__init__()
         self.authenticator = authenticator
 
     @classmethod
@@ -454,6 +476,7 @@ class CredentialsMessage(_MessageType):
     name = 'CREDENTIALS'
 
     def __init__(self, creds):
+        super().__init__()
         self.creds = creds
 
     def send_body(self, f, protocol_version):
@@ -468,11 +491,14 @@ def send_body(self, f, protocol_version):
             write_string(f, credval)
 
 
-class AuthChallengeMessage(_MessageType):
+class AuthChallengeMessage(_DecodableMessageType):
+    __slots__ = ('challenge',)
+
     opcode = 0x0E
     name = 'AUTH_CHALLENGE'
 
     def __init__(self, challenge):
+        super().__init__()
         self.challenge = challenge
 
     @classmethod
@@ -485,17 +511,21 @@ class AuthResponseMessage(_MessageType):
     name = 'AUTH_RESPONSE'
 
     def __init__(self, response):
+        super().__init__()
         self.response = response
 
     def send_body(self, f, protocol_version):
         write_longstring(f, self.response)
 
 
-class AuthSuccessMessage(_MessageType):
+class AuthSuccessMessage(_DecodableMessageType):
+    __slots__ = ('token',)
+
     opcode = 0x10
     name = 'AUTH_SUCCESS'
 
     def __init__(self, token):
+        super().__init__()
         self.token = token
 
     @classmethod
@@ -511,11 +541,14 @@ def send_body(self, f, protocol_version):
         pass
 
 
-class SupportedMessage(_MessageType):
+class SupportedMessage(_DecodableMessageType):
+    __slots__ = ('cql_versions', 'options')
+
     opcode = 0x06
     name = 'SUPPORTED'
 
     def __init__(self, cql_versions, options):
+        super().__init__()
         self.cql_versions = cql_versions
         self.options = options
 
@@ -541,11 +574,15 @@ def recv_body(cls, f, *args):
 
 
 class _QueryMessage(_MessageType):
-
-    def __init__(self, query_params, consistency_level,
-                 serial_consistency_level=None, fetch_size=None,
-                 paging_state=None, timestamp=None, skip_meta=False,
-                 continuous_paging_options=None, keyspace=None):
+    __slots__ = ('query_params', 'consistency_level', 'serial_consistency_level',
+                 'fetch_size', 'paging_state', 'skip_meta', 'timestamp', 'keyspace')
+
+    def __init__(self, query_params, consistency_level, serial_consistency_level=None,
+                 fetch_size=None, paging_state=None, skip_meta=False,
+                 timestamp=None, keyspace=None, continuous_paging_options=None):
+        super().__init__()
+        # Note: continuous_paging_options is accepted for backward compatibility
+        # but is not currently implemented (not stored or used)
         self.query_params = query_params
         self.consistency_level = consistency_level
         self.serial_consistency_level = serial_consistency_level
@@ -607,14 +644,21 @@ def _write_paging_options(self, f, paging_options, protocol_version):
 
 
 class QueryMessage(_QueryMessage):
+    __slots__ = ('query',)
+
     opcode = 0x07
     name = 'QUERY'
 
     def __init__(self, query, consistency_level, serial_consistency_level=None,
                  fetch_size=None, paging_state=None, timestamp=None, continuous_paging_options=None, keyspace=None):
+        # Note: continuous_paging_options is accepted for backward compatibility
+        # but is not currently implemented (not stored or used)
         self.query = query
-        super(QueryMessage, self).__init__(None, consistency_level, serial_consistency_level, fetch_size,
-                                           paging_state, timestamp, False, continuous_paging_options, keyspace)
+        super(QueryMessage, self).__init__(query_params=None, consistency_level=consistency_level,
+                                           serial_consistency_level=serial_consistency_level,
+                                           fetch_size=fetch_size, paging_state=paging_state,
+                                           skip_meta=False, timestamp=timestamp, keyspace=keyspace,
+                                           continuous_paging_options=continuous_paging_options)
 
     def send_body(self, f, protocol_version):
         write_longstring(f, self.query)
@@ -622,6 +666,8 @@ def send_body(self, f, protocol_version):
 
 
 class ExecuteMessage(_QueryMessage):
+    __slots__ = ('query_id', 'result_metadata_id')
+
     opcode = 0x0A
     name = 'EXECUTE'
 
@@ -629,10 +675,15 @@ def __init__(self, query_id, query_params, consistency_level,
                  serial_consistency_level=None, fetch_size=None,
                  paging_state=None, timestamp=None, skip_meta=False,
                  continuous_paging_options=None, result_metadata_id=None):
+        # Note: continuous_paging_options is accepted for backward compatibility
+        # but is not currently implemented (not stored or used)
         self.query_id = query_id
         self.result_metadata_id = result_metadata_id
-        super(ExecuteMessage, self).__init__(query_params, consistency_level, serial_consistency_level, fetch_size,
-                                             paging_state, timestamp, skip_meta, continuous_paging_options)
+        super(ExecuteMessage, self).__init__(query_params=query_params, consistency_level=consistency_level,
+                                             serial_consistency_level=serial_consistency_level,
+                                             fetch_size=fetch_size, paging_state=paging_state,
+                                             skip_meta=skip_meta, timestamp=timestamp, keyspace=None,
+                                             continuous_paging_options=continuous_paging_options)
 
     def _write_query_params(self, f, protocol_version):
         super(ExecuteMessage, self)._write_query_params(f, protocol_version)
@@ -653,14 +704,14 @@ def send_body(self, f, protocol_version):
 RESULT_KIND_SCHEMA_CHANGE = 0x0005
 
 
-class ResultMessage(_MessageType):
+class ResultMessage(_DecodableMessageType):
+    __slots__ = ('kind', 'result_metadata_id', 'results', 'paging_state', 'column_names', 'column_types',
+                 'parsed_rows', 'continuous_paging_seq', 'continuous_paging_last', 'new_keyspace',
+                 'column_metadata', 'query_id', 'bind_metadata', 'pk_indexes', 'schema_change_event', 'is_lwt')
+
     opcode = 0x08
     name = 'RESULT'
 
-    kind = None
-    results = None
-    paging_state = None
-
     # Names match type name in module scope. Most are imported from cassandra.cqltypes (except CUSTOM_TYPE)
     type_codes = _cqltypes_by_code = dict((v, globals()[k]) for k, v in type_codes.__dict__.items() if not k.startswith('_'))
 
@@ -671,25 +722,25 @@ class ResultMessage(_MessageType):
     _CONTINUOUS_PAGING_LAST_FLAG = 0x80000000
     _METADATA_ID_FLAG = 0x0008
 
-    kind = None
-
-    # These are all the things a result message might contain. They are populated according to 'kind'
-    column_names = None
-    column_types = None
-    parsed_rows = None
-    paging_state = None
-    continuous_paging_seq = None
-    continuous_paging_last = None
-    new_keyspace = None
-    column_metadata = None
-    query_id = None
-    bind_metadata = None
-    pk_indexes = None
-    schema_change_event = None
-    is_lwt = False
-
     def __init__(self, kind):
+        super().__init__()
         self.kind = kind
+        # Initialize all slot attributes to None
+        self.result_metadata_id = None
+        self.results = None
+        self.paging_state = None
+        self.column_names = None
+        self.column_types = None
+        self.parsed_rows = None
+        self.continuous_paging_seq = None
+        self.continuous_paging_last = None
+        self.new_keyspace = None
+        self.column_metadata = None
+        self.query_id = None
+        self.bind_metadata = None
+        self.pk_indexes = None
+        self.schema_change_event = None
+        self.is_lwt = None
 
     def recv(self, f, protocol_version, protocol_features, user_type_map, result_metadata, column_encryption_policy):
         if self.kind == RESULT_KIND_VOID:
@@ -859,10 +910,13 @@ def recv_row(f, colcount):
 
 
 class PrepareMessage(_MessageType):
+    __slots__ = ('query', 'keyspace')
+
     opcode = 0x09
     name = 'PREPARE'
 
     def __init__(self, query, keyspace=None):
+        super().__init__()
         self.query = query
         self.keyspace = keyspace
 
@@ -897,12 +951,15 @@ def send_body(self, f, protocol_version):
 
 
 class BatchMessage(_MessageType):
+    __slots__ = ('batch_type', 'queries', 'consistency_level', 'serial_consistency_level',
+                 'timestamp', 'keyspace')
+
     opcode = 0x0D
     name = 'BATCH'
 
     def __init__(self, batch_type, queries, consistency_level,
-                 serial_consistency_level=None, timestamp=None,
-                 keyspace=None):
+                 serial_consistency_level=None, timestamp=None, keyspace=None):
+        super().__init__()
         self.batch_type = batch_type
         self.queries = queries
         self.consistency_level = consistency_level
@@ -962,21 +1019,27 @@ def send_body(self, f, protocol_version):
 
 
 class RegisterMessage(_MessageType):
+    __slots__ = ('event_list',)
+
     opcode = 0x0B
     name = 'REGISTER'
 
     def __init__(self, event_list):
+        super().__init__()
         self.event_list = event_list
 
     def send_body(self, f, protocol_version):
         write_stringlist(f, self.event_list)
 
 
-class EventMessage(_MessageType):
+class EventMessage(_DecodableMessageType):
+    __slots__ = ('event_type', 'event_args')
+
     opcode = 0x0C
     name = 'EVENT'
 
     def __init__(self, event_type, event_args):
+        super().__init__()
         self.event_type = event_type
         self.event_args = event_args
 
@@ -1038,6 +1101,7 @@ class RevisionType(object):
     name = 'REVISE_REQUEST'
 
     def __init__(self, op_type, op_id, next_pages=0):
+        super().__init__()
         self.op_type = op_type
         self.op_id = op_id
         self.next_pages = next_pages
diff --git a/cassandra/query.py b/cassandra/query.py
index 6c6878fdb4..04601e76f2 100644
--- a/cassandra/query.py
+++ b/cassandra/query.py
@@ -214,88 +214,74 @@ class Statement(object):
     An abstract class representing a single query. There are three subclasses:
     :class:`.SimpleStatement`, :class:`.BoundStatement`, and :class:`.BatchStatement`.
     These can be passed to :meth:`.Session.execute()`.
-    """
 
-    retry_policy = None
-    """
-    An instance of a :class:`cassandra.policies.RetryPolicy` or one of its
-    subclasses.  This controls when a query will be retried and how it
-    will be retried.
-    """
+    Attributes
+    ----------
+    retry_policy : RetryPolicy or None
+        An instance of a :class:`cassandra.policies.RetryPolicy` or one of its
+        subclasses.  This controls when a query will be retried and how it
+        will be retried.
 
-    consistency_level = None
-    """
-    The :class:`.ConsistencyLevel` to be used for this operation.  Defaults
-    to :const:`None`, which means that the default consistency level for
-    the Session this is executed in will be used.
-    """
+    consistency_level : ConsistencyLevel or None
+        The :class:`.ConsistencyLevel` to be used for this operation.  Defaults
+        to :const:`None`, which means that the default consistency level for
+        the Session this is executed in will be used.
 
-    fetch_size = FETCH_SIZE_UNSET
-    """
-    How many rows will be fetched at a time.  This overrides the default
-    of :attr:`.Session.default_fetch_size`
+    fetch_size : int
+        How many rows will be fetched at a time.  This overrides the default
+        of :attr:`.Session.default_fetch_size`
 
-    This only takes effect when protocol version 2 or higher is used.
-    See :attr:`.Cluster.protocol_version` for details.
+        This only takes effect when protocol version 2 or higher is used.
+        See :attr:`.Cluster.protocol_version` for details.
 
-    .. versionadded:: 2.0.0
-    """
+        .. versionadded:: 2.0.0
 
-    keyspace = None
-    """
-    The string name of the keyspace this query acts on. This is used when
-    :class:`~.TokenAwarePolicy` is configured in the profile load balancing policy.
+    keyspace : str or None
+        The string name of the keyspace this query acts on. This is used when
+        :class:`~.TokenAwarePolicy` is configured in the profile load balancing policy.
 
-    It is set implicitly on :class:`.BoundStatement`, and :class:`.BatchStatement`,
-    but must be set explicitly on :class:`.SimpleStatement`.
+        It is set implicitly on :class:`.BoundStatement`, and :class:`.BatchStatement`,
+        but must be set explicitly on :class:`.SimpleStatement`.
 
-    .. versionadded:: 2.1.3
-    """
+        .. versionadded:: 2.1.3
 
-    table = None
-    """
-    The string name of the table this query acts on. This is used when the tablet
-    feature is enabled and in the same time :class`~.TokenAwarePolicy` is configured
-    in the profile load balancing policy.
-    """
+    table : str or None
+        The string name of the table this query acts on. This is used when the tablet
+        feature is enabled and in the same time :class`~.TokenAwarePolicy` is configured
+        in the profile load balancing policy.
 
-    custom_payload = None
-    """
-    :ref:`custom_payload` to be passed to the server.
+    custom_payload : dict or None
+        :ref:`custom_payload` to be passed to the server.
 
-    These are only allowed when using protocol version 4 or higher.
+        These are only allowed when using protocol version 4 or higher.
 
-    .. versionadded:: 2.6.0
-    """
+        .. versionadded:: 2.6.0
 
-    is_idempotent = False
-    """
-    Flag indicating whether this statement is safe to run multiple times in speculative execution.
+    is_idempotent : bool
+        Flag indicating whether this statement is safe to run multiple times in speculative execution.
     """
 
-    _serial_consistency_level = None
-    _routing_key = None
+    __slots__ = (
+        'retry_policy', 'consistency_level', 'fetch_size', 'keyspace', 'table',
+        'custom_payload', 'is_idempotent', '_serial_consistency_level', '_routing_key'
+    )
 
     def __init__(self, retry_policy=None, consistency_level=None, routing_key=None,
                  serial_consistency_level=None, fetch_size=FETCH_SIZE_UNSET, keyspace=None, custom_payload=None,
                  is_idempotent=False, table=None):
         if retry_policy and not hasattr(retry_policy, 'on_read_timeout'):  # just checking one method to detect positional parameter errors
             raise ValueError('retry_policy should implement cassandra.policies.RetryPolicy')
-        if retry_policy is not None:
-            self.retry_policy = retry_policy
-        if consistency_level is not None:
-            self.consistency_level = consistency_level
+        # Initialize all attributes (required for __slots__)
+        self.retry_policy = retry_policy
+        self.consistency_level = consistency_level
         self._routing_key = routing_key
+        self._serial_consistency_level = None
         if serial_consistency_level is not None:
             self.serial_consistency_level = serial_consistency_level
-        if fetch_size is not FETCH_SIZE_UNSET:
-            self.fetch_size = fetch_size
-        if keyspace is not None:
-            self.keyspace = keyspace
-        if table is not None:
-            self.table = table
-        if custom_payload is not None:
-            self.custom_payload = custom_payload
+        self.fetch_size = fetch_size
+        self.keyspace = keyspace
+        self.table = table
+        self.custom_payload = custom_payload
         self.is_idempotent = is_idempotent
 
     def _key_parts_packed(self, parts):
@@ -392,6 +378,8 @@ class SimpleStatement(Statement):
     A simple, un-prepared query.
     """
 
+    __slots__ = ('_query_string',)
+
     def __init__(self, query_string, retry_policy=None, consistency_level=None, routing_key=None,
                  serial_consistency_level=None, fetch_size=FETCH_SIZE_UNSET, keyspace=None,
                  custom_payload=None, is_idempotent=False):
@@ -536,18 +524,18 @@ class BoundStatement(Statement):
     """
     A prepared statement that has been bound to a particular set of values.
     These may be created directly or through :meth:`.PreparedStatement.bind()`.
-    """
 
-    prepared_statement = None
-    """
-    The :class:`PreparedStatement` instance that this was created from.
-    """
+    Attributes
+    ----------
+    prepared_statement : PreparedStatement or None
+        The :class:`PreparedStatement` instance that this was created from.
 
-    values = None
-    """
-    The sequence of values that were bound to the prepared statement.
+    values : list or None
+        The sequence of values that were bound to the prepared statement.
     """
 
+    __slots__ = ('prepared_statement', 'values', 'raw_values')
+
     def __init__(self, prepared_statement, retry_policy=None, consistency_level=None, routing_key=None,
                  serial_consistency_level=None, fetch_size=FETCH_SIZE_UNSET, keyspace=None,
                  custom_payload=None):
@@ -557,23 +545,32 @@ def __init__(self, prepared_statement, retry_policy=None, consistency_level=None
         See :class:`Statement` attributes for a description of the other parameters.
         """
         self.prepared_statement = prepared_statement
-
-        self.retry_policy = prepared_statement.retry_policy
-        self.consistency_level = prepared_statement.consistency_level
-        self.serial_consistency_level = prepared_statement.serial_consistency_level
-        self.fetch_size = prepared_statement.fetch_size
-        self.custom_payload = prepared_statement.custom_payload
-        self.is_idempotent = prepared_statement.is_idempotent
         self.values = []
 
+        # Use prepared statement values as defaults if parameters not provided
+        if retry_policy is None:
+            retry_policy = prepared_statement.retry_policy
+        if consistency_level is None:
+            consistency_level = prepared_statement.consistency_level
+        if serial_consistency_level is None:
+            serial_consistency_level = prepared_statement.serial_consistency_level
+        if fetch_size is FETCH_SIZE_UNSET:
+            fetch_size = prepared_statement.fetch_size
+        if custom_payload is None:
+            custom_payload = prepared_statement.custom_payload
+
+        # Get keyspace and table from metadata if available
         meta = prepared_statement.column_metadata
+        table = None
         if meta:
-            self.keyspace = meta[0].keyspace_name
-            self.table = meta[0].table_name
+            if keyspace is None:
+                keyspace = meta[0].keyspace_name
+            table = meta[0].table_name
 
+        # Call parent __init__ with merged parameters
         Statement.__init__(self, retry_policy, consistency_level, routing_key,
                            serial_consistency_level, fetch_size, keyspace, custom_payload,
-                           prepared_statement.is_idempotent)
+                           prepared_statement.is_idempotent, table)
 
     def bind(self, values):
         """
@@ -745,23 +742,19 @@ class BatchStatement(Statement):
     by default.
 
     .. versionadded:: 2.0.0
-    """
 
-    batch_type = None
-    """
-    The :class:`.BatchType` for the batch operation.  Defaults to
-    :attr:`.BatchType.LOGGED`.
-    """
+    Attributes
+    ----------
+    batch_type : BatchType
+        The :class:`.BatchType` for the batch operation.  Defaults to
+        :attr:`.BatchType.LOGGED`.
 
-    serial_consistency_level = None
-    """
-    The same as :attr:`.Statement.serial_consistency_level`, but is only
-    supported when using protocol version 3 or higher.
+    serial_consistency_level : ConsistencyLevel or None
+        The same as :attr:`.Statement.serial_consistency_level`, but is only
+        supported when using protocol version 3 or higher.
     """
 
-    _statements_and_parameters = None
-    _session = None
-    _is_lwt = False
+    __slots__ = ('batch_type', '_statements_and_parameters', '_session', '_is_lwt')
 
     def __init__(self, batch_type=BatchType.LOGGED, retry_policy=None,
                  consistency_level=None, serial_consistency_level=None,
@@ -813,6 +806,7 @@ def __init__(self, batch_type=BatchType.LOGGED, retry_policy=None,
         self.batch_type = batch_type
         self._statements_and_parameters = []
         self._session = session
+        self._is_lwt = False
         Statement.__init__(self, retry_policy=retry_policy, consistency_level=consistency_level,
                            serial_consistency_level=serial_consistency_level, custom_payload=custom_payload)
 
diff --git a/cassandra/tablets.py b/cassandra/tablets.py
index dca26ab0df..6a10affdd9 100644
--- a/cassandra/tablets.py
+++ b/cassandra/tablets.py
@@ -8,10 +8,13 @@ class Tablet(object):
     Represents a single ScyllaDB tablet.
     It stores information about each replica, its host and shard,
     and the token interval in the format (first_token, last_token].
+
+    Attributes:
+        first_token: The start of the token range (exclusive)
+        last_token: The end of the token range (inclusive)
+        replicas: List of replicas for this tablet, each containing (host_id, shard)
     """
-    first_token = 0
-    last_token = 0
-    replicas = None
+    __slots__ = ('first_token', 'last_token', 'replicas')
 
     def __init__(self, first_token=0, last_token=0, replicas=None):
         self.first_token = first_token
@@ -42,8 +45,14 @@ def replica_contains_host_id(self, uuid: UUID) -> bool:
 
 
 class Tablets(object):
-    _lock = None
-    _tablets = {}
+    """
+    Manages the tablet mapping for ScyllaDB tables.
+
+    Attributes:
+        _tablets: Dictionary mapping (keyspace, table) tuples to lists of Tablet objects
+        _lock: Thread lock for synchronizing access to the tablets dictionary
+    """
+    __slots__ = ('_tablets', '_lock')
 
     def __init__(self, tablets):
         self._tablets = tablets
diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py
index 7168ad2940..054f4e9dee 100644
--- a/tests/unit/test_response_future.py
+++ b/tests/unit/test_response_future.py
@@ -16,7 +16,7 @@
 
 from collections import deque
 from threading import RLock
-from unittest.mock import Mock, MagicMock, ANY
+from unittest.mock import Mock, MagicMock, ANY, patch
 
 from cassandra import ConsistencyLevel, Unavailable, SchemaTargetType, SchemaChangeType, OperationTimedOut
 from cassandra.cluster import Session, ResponseFuture, NoHostAvailable, ProtocolVersion
@@ -632,15 +632,16 @@ def test_repeat_orig_query_after_succesful_reprepare(self):
         response.results = (None, None, None, None, None)
         response.query_id = query_id
 
-        rf._query = Mock(return_value=True)
-        rf._execute_after_prepare('host', None, None, response)
-        rf._query.assert_called_once_with('host')
+        with patch.object(ResponseFuture, '_query', return_value=True) as mock_query:
+            rf._execute_after_prepare('host', None, None, response)
+            # When patching at class level, self is not passed to the mock
+            mock_query.assert_called_once_with('host')
 
         rf.prepared_statement = Mock()
         rf.prepared_statement.query_id = query_id
-        rf._query = Mock(return_value=True)
-        rf._execute_after_prepare('host', None, None, response)
-        rf._query.assert_called_once_with('host')
+        with patch.object(ResponseFuture, '_query', return_value=True) as mock_query:
+            rf._execute_after_prepare('host', None, None, response)
+            mock_query.assert_called_once_with('host')
 
     def test_timeout_does_not_release_stream_id(self):
         """