diff --git a/CHANGELOG.md b/CHANGELOG.md index 1125a354..498b19d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,8 @@ ### 0.3.2 -- Restore %(param)s style param substition support. By @baconfield in #147 +- Fix connection secure parameter management. By @stankudrow and @barakor in #162 (inspired by PR #138 by @barakor). +- Restore %(param)s style param substition support. By @baconfield in #147. ### 0.3.1 diff --git a/NOTES.md b/NOTES.md new file mode 100644 index 00000000..76312d22 --- /dev/null +++ b/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. diff --git a/README.md b/README.md index a46021d1..330743fd 100644 --- a/README.md +++ b/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(): diff --git a/asynch/connection.py b/asynch/connection.py index 27383dcc..a1002aa0 100644 --- a/asynch/connection.py +++ b/asynch/connection.py @@ -1,9 +1,10 @@ -from typing import Optional +from typing import Any, Dict, Optional from asynch.cursors import Cursor from asynch.errors import NotSupportedError from asynch.proto import constants from asynch.proto.connection import Connection as ProtoConnection +from asynch.proto.connection import get_default_port from asynch.proto.models.enums import ConnectionStatus from asynch.proto.utils.dsn import parse_dsn @@ -12,47 +13,72 @@ class Connection: def __init__( self, dsn: Optional[str] = None, - user: str = constants.DEFAULT_USER, - password: str = constants.DEFAULT_PASSWORD, - host: str = constants.DEFAULT_HOST, - port: int = constants.DEFAULT_PORT, - database: str = constants.DEFAULT_DATABASE, + user: Optional[str] = None, + password: Optional[str] = None, + host: Optional[str] = None, + port: Optional[int] = None, + database: Optional[str] = None, cursor_cls=Cursor, echo: bool = False, stack_track: bool = False, + secure: bool = False, **kwargs, ): - 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 - else: - self._connection = ProtoConnection( - host=host, - port=port, - database=database, - user=user, - password=password, - stack_track=stack_track, - **kwargs, - ) - self._dsn = dsn - # dsn parts - self._user = user - self._password = password - self._host = host - self._port = port - self._database = database + self._proto_kwargs = self._get_proto_connection_kwargs( + dsn, + user, + password, + host, + port, + database, + secure=secure, + stack_track=stack_track, + **kwargs, + ) + self._connection = ProtoConnection(**self._proto_kwargs) + self._cursor_cls = cursor_cls + self._echo = echo # connection additional settings self._opened: bool = False self._closed: bool = False - self._cursor_cls = cursor_cls - self._connection_kwargs = kwargs - self._echo = echo + + @staticmethod + def _get_proto_connection_kwargs( + dsn: Optional[str], + user: Optional[str], + password: Optional[str], + host: Optional[str], + port: Optional[int], + database: Optional[str], + *, + secure: bool, + stack_track: bool, + **kwargs, + ) -> Dict[str, Any]: + conninfo: Dict[str, Any] = {} + if dsn: + config = parse_dsn(dsn) + conninfo = {**conninfo, **config} + # Parameters suppercede DSN parts! + # If a param is given and it is not False-evaliuated, this is it, + # otherwise the corresponding DSN part is used OR the param itself. + user = user or config.get("user", constants.DEFAULT_USER) + password = password or config.get("password", constants.DEFAULT_PASSWORD) + host = host or config.get("host", constants.DEFAULT_HOST) + port = port or config.get("port", get_default_port(secure=secure)) + database = database or config.get("database", constants.DEFAULT_DATABASE) + secure = secure or config.get("secure", secure) + return { + **conninfo, + "user": user, + "password": password, + "host": host, + "port": port, + "database": database, + "secure": secure, + "stack_track": stack_track, + **kwargs, + } async def __aenter__(self) -> "Connection": await self.connect() @@ -104,24 +130,24 @@ def status(self) -> str: raise ConnectionError(f"{self} is in an unknown state") @property - def host(self) -> str: - return self._host + def host(self) -> Optional[str]: + return self._proto_kwargs.get("host") @property - def port(self) -> int: - return self._port + def port(self) -> Optional[int]: + return self._proto_kwargs.get("port") @property - def user(self) -> str: - return self._user + def user(self) -> Optional[str]: + return self._proto_kwargs.get("user") @property - def password(self) -> str: - return self._password + def password(self) -> Optional[str]: + return self._proto_kwargs.get("password") @property - def database(self) -> str: - return self._database + def database(self) -> Optional[str]: + return self._proto_kwargs.get("database") @property def echo(self) -> bool: diff --git a/asynch/pool.py b/asynch/pool.py index 7aa1f381..31d4560b 100644 --- a/asynch/pool.py +++ b/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 diff --git a/asynch/proto/connection.py b/asynch/proto/connection.py index 1c2f3070..4fa7fa69 100644 --- a/asynch/proto/connection.py +++ b/asynch/proto/connection.py @@ -41,6 +41,19 @@ logger = logging.getLogger(__name__) +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 + + class QueryProcessingStage: """Determines till which state SELECT query should be executed.""" @@ -77,7 +90,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 +104,18 @@ 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, ): + port = get_default_port(secure=secure) if port is None else port 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 diff --git a/asynch/proto/utils/dsn.py b/asynch/proto/utils/dsn.py index 3087cf2f..a6946066 100644 --- a/asynch/proto/utils/dsn.py +++ b/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] """ diff --git a/asynch/proto/utils/escape.py b/asynch/proto/utils/escape.py index c7dfe4fe..1afe60a6 100644 --- a/asynch/proto/utils/escape.py +++ b/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) diff --git a/example.env b/example.env index 99cf60ce..5259339e 100644 --- a/example.env +++ b/example.env @@ -1,5 +1,5 @@ CLICKHOUSE_HOST=localhost CLICKHOUSE_PORT=9000 -CLICKHOUSE_USER=default -CLICKHOUSE_PASSWORD=default -CLICKHOUSE_DB=default \ No newline at end of file +CLICKHOUSE_USER=ch +CLICKHOUSE_PASSWORD=asynch +CLICKHOUSE_DB=test diff --git a/tests/test_connection.py b/tests/test_connection.py index 6810670f..4f7668aa 100644 --- a/tests/test_connection.py +++ b/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 @@ -41,7 +42,7 @@ def _test_connectivity_invariant( assert conn.closed is is_closed -def test_dsn(): +def test_dsn() -> None: dsn = f"clickhouse://{USER}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}" conn = Connection(dsn=dsn) @@ -51,6 +52,60 @@ def test_dsn(): _test_connectivity_invariant(conn=conn) +@pytest.mark.parametrize( + ("dsn", "params", "answer"), + [ + ( + "", + { + "host": HOST, + "port": PORT, + "user": USER, + "password": PASSWORD, + "database": DATABASE, + }, + { + "host": HOST, + "port": PORT, + "user": USER, + "password": PASSWORD, + "database": DATABASE, + }, + ), + ( + f"clickhouses://{USER}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}", + { + "host": "243.164.23.54", + "port": None, + "user": "test_user", + "database": "db", + }, + { + "host": "243.164.23.54", + "port": PORT, + "user": "test_user", + "password": PASSWORD, + "database": "db", + }, + ), + ( + f"clickhouses://{USER}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}", + {}, + { + "host": HOST, + "port": PORT, + "user": USER, + "password": PASSWORD, + "database": DATABASE, + }, + ), + ], +) +def test_connection_params(dsn: str, params: dict, answer: dict) -> None: + conn = Connection(dsn=dsn, **params) + _test_connection_credentials(conn, **answer) + + def test_secure_dsn(): dsn = ( f"clickhouses://{USER}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}" @@ -134,7 +189,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 +213,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 +233,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 +265,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,7 +286,7 @@ 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() @@ -239,7 +294,7 @@ async def test_connection_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() diff --git a/tests/test_cursors.py b/tests/test_cursors.py index c5288452..642e4754 100644 --- a/tests/test_cursors.py +++ b/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}] diff --git a/tests/test_pool.py b/tests/test_pool.py index b392ea1c..98e842b2 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -8,6 +8,7 @@ from asynch.pool import Pool from asynch.proto import constants from asynch.proto.models.enums import PoolStatus +from tests.conftest import CONNECTION_DSN def _get_pool_size(pool: Pool) -> int: @@ -41,7 +42,7 @@ async def test_pool_repr(): assert repr(pool) == repstr min_size, max_size = 2, 3 - pool = Pool(minsize=min_size, maxsize=max_size) + pool = Pool(minsize=min_size, maxsize=max_size, dsn=CONNECTION_DSN) async with pool: repstr = ( f"