bgunebakan · Oct 16, 2026 · Oct 15, 2026 · Oct 15, 2026 · Oct 15, 2026 · Oct 15, 2026
diff --git a/pyproject.toml b/pyproject.toml
@@ -174,7 +174,7 @@ lint.per-file-ignores."tests/*" = [
     "S310", # False positive; it's a https url
     "S311", # Standard pseudo-random generators are not suitable for cryptographic purposes
 ]
-lint.per-file-ignores."src/crate/client/{connection.py,http.py}" = [
+lint.per-file-ignores."src/crate/client/{connection.py,http.py,_pool.py,_response.py}" = [
     "A004", # Import `ConnectionError` is shadowing a Python builtin
     "A005", # Import `ConnectionError` is shadowing a Python builtin
 ]

diff --git a/src/crate/client/_codec.py b/src/crate/client/_codec.py
@@ -0,0 +1,79 @@
+# -*- coding: utf-8; -*-
+#
+# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# with Crate these terms will supersede the license and you may use the
+# software solely pursuant to the terms of the relevant commercial agreement.
+
+import calendar
+import datetime as dt
+import typing as t
+from decimal import Decimal
+
+import orjson
+
+epoch_aware = dt.datetime(1970, 1, 1, tzinfo=dt.timezone.utc)
+epoch_naive = dt.datetime(1970, 1, 1)
+
+
+def json_encoder(obj: t.Any) -> t.Union[int, str]:
+    """
+    Encoder function for orjson, with additional type support.
+
+    - Python's `Decimal` type will be serialized to `str`.
+    - Python's `dt.datetime` and `dt.date` types will be
+      serialized to `int` after converting to milliseconds
+      since epoch.
+    - Python's `dt.time` will be serialized to `str`, following
+    the ISO format.
+
+    https://github.com/ijl/orjson#default
+    https://cratedb.com/docs/crate/reference/en/latest/general/ddl/data-types.html#type-timestamp
+    """
+    if isinstance(obj, Decimal):
+        return str(obj)
+    if isinstance(obj, dt.datetime):
+        if obj.tzinfo is not None:
+            delta = obj - epoch_aware
+        else:
+            delta = obj - epoch_naive
+        return int(
+            delta.microseconds / 1000.0
+            + (delta.seconds + delta.days * 24 * 3600) * 1000.0
+        )
+    if isinstance(obj, dt.time):
+        return obj.isoformat()
+    if isinstance(obj, dt.date):
+        return calendar.timegm(obj.timetuple()) * 1000
+    raise TypeError
+
+
+def json_dumps(obj: t.Any) -> bytes:
+    """
+    Serialize to JSON format, using `orjson`, with additional type support.
+
+    https://github.com/ijl/orjson
+    """
+    return orjson.dumps(
+        obj,
+        default=json_encoder,
+        option=(
+            orjson.OPT_PASSTHROUGH_DATETIME
+            | orjson.OPT_NON_STR_KEYS
+            | orjson.OPT_SERIALIZE_NUMPY
+        ),
+    )
diff --git a/src/crate/client/_options.py b/src/crate/client/_options.py
@@ -0,0 +1,120 @@
+# -*- coding: utf-8; -*-
+#
+# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# with Crate these terms will supersede the license and you may use the
+# software solely pursuant to the terms of the relevant commercial agreement.
+
+import re
+import socket
+import typing as t
+from urllib.parse import urlparse
+
+SQL_PATH = "/_sql?types=true"
+
+_HTTP_PAT = re.compile("https?://.+", re.I)
+
+
+def server_url(server: str) -> str:
+    """
+    Normalizes a given server string to an url
+
+    >>> print(server_url('a'))
+    http://a
+    >>> print(server_url('a/path'))
+    http://a/path
+    >>> print(server_url('a:9345'))
+    http://a:9345
+    >>> print(server_url('a:9345/path'))
+    http://a:9345/path
+    >>> print(server_url('https://a:9345'))
+    https://a:9345
+    >>> print(server_url('https://a:9345/path'))
+    https://a:9345/path
+    >>> print(server_url('https://a'))
+    https://a
+    >>> print(server_url('https://a/path'))
+    https://a/path
+    >>> print(server_url('demo.crate.io'))
+    http://demo.crate.io
+    """
+    if not _HTTP_PAT.match(server):
+        server = "http://%s" % server
+    parsed = urlparse(server)
+    return parsed.geturl()
+
+
+def to_server_list(servers: t.Union[str, t.Iterable[str]]) -> t.List[str]:
+    if isinstance(servers, str):
+        servers = servers.split()
+    return [server_url(s) for s in servers]
+
+
+def credentials_from_url(
+    url: str, username: t.Optional[str], password: t.Optional[str]
+) -> t.Tuple[t.Optional[str], t.Optional[str]]:
+    """
+    Return the user name and password from the userinfo of `url`, falling
+    back to the given ones for the parts the URL does not contain.
+    """
+    parsed = urlparse(url)
+    if parsed.username is not None:
+        username = parsed.username
+    if parsed.password is not None:
+        password = parsed.password
+    return username, password
+
+
+def validate_compress(compress: t.Any) -> t.Union[int, bool]:
+    if not isinstance(compress, (bool, int)):
+        raise TypeError(
+            f"compress must be bool or int, got {type(compress).__name__!r}"
+        )
+    return compress
+
+
+def sql_path(error_trace: bool, path: str = SQL_PATH) -> str:
+    if error_trace:
+        path += "&error_trace=true"
+    return path
+
+
+def keepalive_socket_options(
+    keepalive: bool = True,
+    tcp_keepidle: t.Optional[int] = None,
+    tcp_keepintvl: t.Optional[int] = None,
+    tcp_keepcnt: t.Optional[int] = None,
+) -> t.List[t.Tuple[int, int, int]]:
+    """
+    Return the `setsockopt` arguments for TCP keepalive, or an empty list
+    when `keepalive` is off.
+    """
+    if not keepalive:
+        return []
+
+    # always use TCP keepalive
+    opts = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)]
+
+    # hasattr check because some options depend on system capabilities
+    # see https://docs.python.org/3/library/socket.html#socket.SOMAXCONN
+    if hasattr(socket, "TCP_KEEPIDLE") and tcp_keepidle is not None:
+        opts.append((socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, tcp_keepidle))
+    if hasattr(socket, "TCP_KEEPINTVL") and tcp_keepintvl is not None:
+        opts.append((socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, tcp_keepintvl))
+    if hasattr(socket, "TCP_KEEPCNT") and tcp_keepcnt is not None:
+        opts.append((socket.IPPROTO_TCP, socket.TCP_KEEPCNT, tcp_keepcnt))
+    return opts
diff --git a/src/crate/client/_pool.py b/src/crate/client/_pool.py
@@ -0,0 +1,120 @@
+# -*- coding: utf-8; -*-
+#
+# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# with Crate these terms will supersede the license and you may use the
+# software solely pursuant to the terms of the relevant commercial agreement.
+
+import heapq
+import logging
+import threading
+import time
+import typing as t
+
+from crate.client.exceptions import ConnectionError
+
+
+class _ServerPool:
+    """
+    Round-robin server selection. A dropped server is taken out of rotation
+    for `retry_interval` seconds.
+
+    Subclasses set `_logger` to log server changes on their own logger.
+    """
+
+    retry_interval = 30
+    """Retry interval for failed servers in seconds."""
+
+    _logger = logging.getLogger(__name__)
+
+    def __init__(
+        self,
+        servers: t.List[str],
+        clock: t.Callable[[], float] = time.time,
+    ) -> None:
+        self._active_servers = servers
+        self._inactive_servers: t.List[t.Tuple[float, str, str]] = []
+        self._lock = threading.RLock()
+        self._clock = clock
+
+    def _get_server(self) -> str:
+        """
+        Get server to use for request.
+        Also process inactive server list, re-add them after given interval.
+        """
+        with self._lock:
+            inactive_server_count = len(self._inactive_servers)
+            for _ in range(inactive_server_count):
+                try:
+                    ts, server, message = heapq.heappop(self._inactive_servers)
+                except IndexError:
+                    pass
+                else:
+                    if (ts + self.retry_interval) > self._clock():
+                        # Not yet, put it back
+                        heapq.heappush(
+                            self._inactive_servers, (ts, server, message)
+                        )
+                    else:
+                        self._active_servers.append(server)
+                        self._logger.warning(
+                            "Restored server %s into active pool", server
+                        )
+
+            # if none is old enough, use oldest
+            if not self._active_servers:
+                ts, server, message = heapq.heappop(self._inactive_servers)
+                self._active_servers.append(server)
+                self._logger.info("Restored server %s into active pool", server)
+
+            server = self._active_servers[0]
+            self._roundrobin()
+
+            return server
+
+    @property
+    def active_servers(self) -> t.List[str]:
+        """get the active servers for this client"""
+        with self._lock:
+            return list(self._active_servers)
+
+    def _drop_server(self, server: str, message: str) -> None:
+        """
+        Drop server from active list and adds it to the inactive ones.
+        """
+        try:
+            self._active_servers.remove(server)
+        except ValueError:
+            pass
+        else:
+            heapq.heappush(
+                self._inactive_servers, (self._clock(), server, message)
+            )
+            self._logger.warning("Removed server %s from active pool", server)
+
+        # if this is the last server raise exception, otherwise try next
+        if not self._active_servers:
+            raise ConnectionError(
+                ("No more Servers available, exception from last server: %s")
+                % message
+            )
+
+    def _roundrobin(self) -> None:
+        """
+        Very simple round-robin implementation
+        """
+        self._active_servers.append(self._active_servers.pop(0))