Fix connection secure port management.

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### 0.3.2
 
 - Restore %(param)s style param substition support. By @baconfield in #147
+- Fix connection secure parameter management. By @stankudrow in #162 (inspired by PR #138 by @barakor)
 
 ### 0.3.1
 

NOTES.md

@@ -0,0 +1,15 @@
+# Project notes
+
+## Table of contents
+
+- [How to test](#how-to-test)
+
+### How to test
+
+A way (but not the sole one) to do it:
+
+1. Make sure the virtual environment (venv) is activated (use `poetry shell` to activate it if necessary);
+2. Create a `.env` file if necessary (you can copypaste the contents of the [example.env](./example.env) file);
+3. Inject the environment variable from the `.env` file into the virtual environment -> on Ubuntu you can do it with `export $(cat .env | xargs)`;
+4. Run a "clickhouse-server" container in a terminal. For example, `docker run --rm --name clickhouse-server -p 8123:8123 -p 9000:9000 --ulimit nofile=262144:262144 -v clickhouse_data:/var/lib/clickhouse -v clickhouse_logs:/var/log/clickhouse-server -e CLICKHOUSE_USER=ch -e CLICKHOUSE_PASSWORD=asynch -e CLICKHOUSE_DB=test clickhouse/clickhouse-server`;
+5. Launch `make test` in another terminal.

README.md

@@ -230,7 +230,7 @@ async def use_pool():
                 assert ret == (1,)
 ```
 
-Or, you may opne/close the pool manually:
+Or, you may open/close the pool manually:
 
 ```python
 async def use_pool():

asynch/connection.py

@@ -6,6 +6,7 @@
 from asynch.proto.connection import Connection as ProtoConnection
 from asynch.proto.models.enums import ConnectionStatus
 from asynch.proto.utils.dsn import parse_dsn
+from asynch.proto.utils.helpers import get_default_port
 
 
 class Connection:
@@ -15,21 +16,26 @@ def __init__(
         user: str = constants.DEFAULT_USER,
         password: str = constants.DEFAULT_PASSWORD,
         host: str = constants.DEFAULT_HOST,
-        port: int = constants.DEFAULT_PORT,
+        port: Optional[int] = None,
         database: str = constants.DEFAULT_DATABASE,
         cursor_cls=Cursor,
         echo: bool = False,
         stack_track: bool = False,
+        secure: bool = False,
         **kwargs,
     ):
+        self._dsn = dsn
+        if port is None:
+            port = get_default_port(secure=secure)
         if dsn:
             config = parse_dsn(dsn)
             self._connection = ProtoConnection(**config, stack_track=stack_track, **kwargs)
-            user = config.get("user", None) or user
-            password = config.get("password", None) or password
-            host = config.get("host", None) or host
-            port = config.get("port", None) or port
-            database = config.get("database", None) or database
+            self._user = config.get("user")
+            self._password = config.get("password")
+            self._host = config.get("host")
+            self._port = config.get("port")
+            self._database = config.get("database")
+            self._secure = config.get("secure")
         else:
             self._connection = ProtoConnection(
                 host=host,
@@ -38,15 +44,15 @@ def __init__(
                 user=user,
                 password=password,
                 stack_track=stack_track,
+                secure=secure,
                 **kwargs,
             )
-        self._dsn = dsn
-        # dsn parts
-        self._user = user
-        self._password = password
-        self._host = host
-        self._port = port
-        self._database = database
+            self._user = user
+            self._password = password
+            self._host = host
+            self._port = port
+            self._database = database
+            self._secure = secure
         # connection additional settings
         self._opened: bool = False
         self._closed: bool = False
@@ -104,23 +110,23 @@ def status(self) -> str:
         raise ConnectionError(f"{self} is in an unknown state")
 
     @property
-    def host(self) -> str:
+    def host(self) -> Optional[str]:
         return self._host
 
     @property
-    def port(self) -> int:
+    def port(self) -> Optional[int]:
         return self._port
 
     @property
-    def user(self) -> str:
+    def user(self) -> Optional[str]:
         return self._user
 
     @property
-    def password(self) -> str:
+    def password(self) -> Optional[str]:
         return self._password
 
     @property
-    def database(self) -> str:
+    def database(self) -> Optional[str]:
         return self._database
 
     @property

asynch/pool.py

@@ -190,7 +190,7 @@ async def _init_connections(self, n: int, *, strict: bool = False) -> None:
             raise ValueError(msg)
         if (self._pool_size + n) > self.maxsize:
             msg = (
-                f"{self} has the {self._pool_size} connections, "
+                f"{self} has {self._pool_size} connections, "
                 f"adding {n} will exceed its maxsize ({self.maxsize})"
             )
             raise AsynchPoolError(msg)
@@ -207,7 +207,7 @@ async def _init_connections(self, n: int, *, strict: bool = False) -> None:
             for i in await asyncio.gather(*tasks, return_exceptions=True)
             if isinstance(i, Exception)
         ):
-            msg = f"failed to create the {n} connection(s) for the {self}"
+            msg = f"failed to create {n} connection(s) for {self}"
             raise AsynchPoolError(msg)
 
     async def _ensure_minsize_connections(self, *, strict: bool = False) -> None:
@@ -243,9 +243,7 @@ async def connection(self) -> AsyncIterator[Connection]:
     async def startup(self) -> "Pool":
         """Initialise the pool.
 
-        When entering the context,
-        the pool get filled with connections
-        up to the pool `minsize` value.
+        The pool is filled with `minsize` connections.
 
         :return: a pool object with `minsize` opened connections
         :rtype: Pool

asynch/proto/connection.py

@@ -36,7 +36,7 @@
 from asynch.proto.streams.block import BlockReader, BlockWriter
 from asynch.proto.streams.buffered import BufferedReader, BufferedWriter
 from asynch.proto.utils.escape import escape_params
-from asynch.proto.utils.helpers import chunks, column_chunks
+from asynch.proto.utils.helpers import chunks, column_chunks, get_default_port
 
 logger = logging.getLogger(__name__)
 
@@ -77,7 +77,7 @@ def __init__(  # nosec:B107
         user: str = constants.DEFAULT_USER,
         password: str = constants.DEFAULT_PASSWORD,
         host: str = constants.DEFAULT_HOST,
-        port: int = constants.DEFAULT_PORT,
+        port: Optional[int] = None,
         database: str = constants.DEFAULT_DATABASE,
         client_name: str = constants.CLIENT_NAME,
         connect_timeout: int = constants.DBMS_DEFAULT_CONNECT_TIMEOUT_SEC,
@@ -91,21 +91,19 @@ def __init__(  # nosec:B107
         ssl_version=None,
         ca_certs=None,
         ciphers=None,
-        alt_hosts: str = None,
+        alt_hosts: str = "",
         stack_track=False,
         settings_is_important=False,
         **kwargs,
     ):
+        if port is None:
+            port = get_default_port(secure=secure)
         self.stack_track = stack_track
-        if secure:
-            default_port = constants.DEFAULT_SECURE_PORT
-        else:
-            default_port = constants.DEFAULT_PORT
-        self.hosts = [(host, port or default_port)]
+        self.hosts = [(host, port)]
         if alt_hosts:
             for host in alt_hosts.split(","):
                 url = urlparse(f"{ClickhouseScheme.clickhouse}://" + host)
-                self.hosts.append((url.hostname, url.port or default_port))
+                self.hosts.append((url.hostname or host, url.port or port))
         self.database = database
         self.host = None
         self.port = None

asynch/proto/utils/dsn.py

@@ -16,8 +16,7 @@
 _TIMEOUTS: set[str] = {"connect_timeout", "send_receive_timeout", "sync_request_timeout"}
 
 
-class DSNError(Exception):
-    pass
+class DSNError(Exception): ...
 
 
 def parse_dsn(dsn: str) -> dict[str, Any]:
@@ -33,9 +32,9 @@ def parse_dsn(dsn: str) -> dict[str, Any]:
 
     :param dsn str: the DSN string
 
-    :raises DSNError: when parsing fails under the strict mode
+    :raises DSNError: DNS parsing failure(s)
 
-    :return: the dictionary of DSN string components
+    :return: DSN configuration dictionary
     :rtype: dict[str, Any]
     """
 

asynch/proto/utils/escape.py

@@ -22,28 +22,20 @@
 def escape_param(item: Any) -> str:
     if item is None:
         return "NULL"
-
     elif isinstance(item, datetime):
         return "'%s'" % item.strftime("%Y-%m-%d %H:%M:%S")
-
     elif isinstance(item, date):
         return "'%s'" % item.strftime("%Y-%m-%d")
-
     elif isinstance(item, string_types):
         return "'%s'" % "".join(escape_chars_map.get(c, c) for c in item)
-
     elif isinstance(item, list):
         return "[%s]" % ", ".join(text_type(escape_param(x)) for x in item)
-
     elif isinstance(item, tuple):
         return "(%s)" % ", ".join(text_type(escape_param(x)) for x in item)
-
     elif isinstance(item, Enum):
         return escape_param(item.value)
-
     elif isinstance(item, UUID):
         return "'%s'" % str(item)
-
     else:
         return str(item)
 

asynch/proto/utils/helpers.py

@@ -1,5 +1,20 @@
 from itertools import islice, tee
 
+from asynch.proto import constants
+
+
+def get_default_port(*, secure: bool = False) -> int:
+    """Returns the default port.
+
+    :param secure bool: return the default secure port
+
+    :returns: the default port
+    :rtype: int
+    """
+    if secure:
+        return constants.DEFAULT_SECURE_PORT
+    return constants.DEFAULT_PORT
+
 
 def chunks(seq, n):
     it = iter(seq)

example.env

@@ -1,5 +1,5 @@
 CLICKHOUSE_HOST=localhost
 CLICKHOUSE_PORT=9000
-CLICKHOUSE_USER=default
-CLICKHOUSE_PASSWORD=default
-CLICKHOUSE_DB=default
+CLICKHOUSE_USER=ch
+CLICKHOUSE_PASSWORD=asynch
+CLICKHOUSE_DB=test

tests/test_connection.py

@@ -3,6 +3,7 @@
 import pytest
 
 from asynch.connection import Connection
+from tests.conftest import CONNECTION_DSN
 
 HOST = "192.168.15.103"
 PORT = 10000
@@ -134,7 +135,7 @@ def test_connection_status_offline():
 
 @pytest.mark.asyncio
 async def test_connection_status_online():
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
     conn_id = id(conn)
 
     repstr = f"<{conn.__class__.__name__} object at 0x{conn_id:x}"
@@ -158,7 +159,7 @@ async def test_connection_status_online():
 
 @pytest.mark.asyncio
 async def test_async_context_manager_interface():
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
     _test_connectivity_invariant(conn=conn)
 
     async with conn:
@@ -178,7 +179,7 @@ async def test_async_context_manager_interface():
 
 @pytest.mark.asyncio
 async def test_connection_ping():
-    conn = Connection()  # default
+    conn = Connection(dsn=CONNECTION_DSN)
 
     with pytest.raises(ConnectionError):
         await conn.ping()
@@ -210,13 +211,13 @@ async def test_connection_cleanup(get_tcp_connections):
 
     # get the number of total TCP connections to the ClickHouse
     init_tcps = 0
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
     async with conn as cn:
         init_tcps = await get_tcp_connections(cn)
 
     # open-execute-close connections
     for _ in range(100):
-        async with Connection() as cn:
+        async with Connection(dsn=CONNECTION_DSN) as cn:
             async with cn.cursor() as cur:
                 await cur.execute("SELECT 1")
                 ret = await cur.fetchone()
@@ -231,15 +232,15 @@ async def test_connection_cleanup(get_tcp_connections):
 
 @pytest.mark.asyncio
 async def test_connection_close():
-    conn = Connection()
+    conn = Connection(dsn=CONNECTION_DSN)
 
     # it does not break
     await conn.close()
 
     assert not conn.opened
     assert conn.closed
 
-    async with Connection() as conn:
+    async with Connection(dsn=CONNECTION_DSN) as conn:
         assert conn.opened
 
         await conn.close()

tests/test_cursors.py

@@ -58,7 +58,7 @@ async def test_fetchone(conn: Connection):
         ret = await cursor.fetchone()
         assert ret == (1,)
 
-        await cursor.execute("SELECT {val}", args={"val": 2})
+        await cursor.execute("SELECT %(val)s", args={"val": 2})
         ret = await cursor.fetchone()
         assert ret == (2,)
 
@@ -74,7 +74,7 @@ async def test_fetchall(conn: Connection):
         ret = await cursor.fetchall()
         assert ret == [(1,)]
 
-        await cursor.execute("SELECT {val}", args={"val": 2})
+        await cursor.execute("SELECT %(val)s", args={"val": 2})
         ret = await cursor.fetchall()
         assert ret == [(2,)]
 
@@ -86,7 +86,7 @@ async def test_dict_cursor(conn: Connection):
         ret = await cursor.fetchall()
         assert ret == [{"1": 1}]
 
-        await cursor.execute("SELECT {val}", args={"val": 2})
+        await cursor.execute("SELECT %(val)s", args={"val": 2})
         ret = await cursor.fetchall()
         assert ret == [{"2": 2}]