From 235d15f3abce859c7667f1fc0037188e002300c1 Mon Sep 17 00:00:00 2001 From: Andy Mc Donald Date: Tue, 10 Mar 2026 14:10:41 +0000 Subject: [PATCH] 3.1.8 --- docs/release-notes/changelog.md | 15 ++++- mkdocs.yml | 2 +- src/pykx/db.py | 11 +++- src/pykx/ipc.py | 57 ++++++++++++++--- src/pykx/pykxq_m.c | 2 +- tests/test_db.py | 71 +++++++++++++++++++++ tests/test_ipc.py | 107 ++++++++++++++++++++++++++++++++ 7 files changed, 251 insertions(+), 14 deletions(-) diff --git a/docs/release-notes/changelog.md b/docs/release-notes/changelog.md index 9766bc3..e2dad40 100644 --- a/docs/release-notes/changelog.md +++ b/docs/release-notes/changelog.md @@ -4,11 +4,22 @@ The changelog presented here outlines changes to PyKX when operating within a Python environment specifically, if you require changelogs associated with PyKX operating under a q environment see [here](./underq-changelog.md). -# PyKX 3.1.7 +# PyKX 3.1.8 #### Release Date -2026-02-09 +2026-02-18 + +### Additions + +- Added `connection_timeout` keyword argument to `AsyncQConnection` objects allowing the specification of a timeout for the initial connection to a `q` server. +- Added the usage of `.q.Q.ens` for `sym_enum` argument in `DB` objects. + +## PyKX 3.1.7 + +#### Release Date + +2026-02-10 ### Additions diff --git a/mkdocs.yml b/mkdocs.yml index ef490ea..ab75569 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -283,7 +283,7 @@ nav: - PyKX as a server: examples/server/server.md - Real-time streaming: examples/streaming/index.md - Multithreaded execution: examples/threaded_execution/threading.md - - Asyncronous Queries: examples/AsynchronousQueries/async_querying.md + - Asynchronous Queries: examples/AsynchronousQueries/async_querying.md - Releases: - Release notes: - PyKX: release-notes/changelog.md diff --git a/src/pykx/db.py b/src/pykx/db.py index 2458a50..a021920 100644 --- a/src/pykx/db.py +++ b/src/pykx/db.py @@ -146,7 +146,13 @@ def __init__(self, >>> db.tables ['quote', 'trade'] ``` + !!! note "Loading root-level objects" + When using `pykx.DB`, only partitioned tables are exposed as attributes of the + `DB` object. Root-level splayed objects (for example, dictionaries or lists stored + alongside partitions) are **not** attached to the `DB` instance, but are still loaded + into the q process and can be accessed directly using `kx.q`. + For example, you can retrieve a variable `mySplayTab` by running `kx.q['mySplayTab']`. Define the path to be used for a database which does not initially exist ```python @@ -323,7 +329,10 @@ def create(self, qfunc = q(_func_mapping[func_name]) try: if format == 'splayed': - table = q.Q.en(save_dir, table) + if sym_enum is None: + table = q.Q.en(save_dir, table) + else: + table = q.Q.ens(save_dir, table, sym_enum) q('{.Q.dd[x;`] set y}', save_dir/table_name, table) else: if isinstance(partition, str): diff --git a/src/pykx/ipc.py b/src/pykx/ipc.py index fb60e69..fca6b8f 100644 --- a/src/pykx/ipc.py +++ b/src/pykx/ipc.py @@ -458,7 +458,9 @@ def __init__(self, no_ctx: bool = False, reconnection_attempts: int = -1, reconnection_delay: float = 0.5, - reconnection_function: callable = reconnection_function + reconnection_function: callable = reconnection_function, + connection_timeout: Optional[float] = None, + ): """Interface with a q process using the q IPC protocol. @@ -498,6 +500,9 @@ def __init__(self, reconnection_delay: A `#!python float` for the initial delay between reconnect attempts (in seconds). This is passed to the provided `#!python reconnection_function` that is executed on reconnect attempt. + connection_timeout: Timeout in seconds for connection to `q` server. If left as default + `None`, the socket will not timeout when connecting. + Note: The `#!python username` and `#!python password` parameters are not required. The `#!python username` and `#!python password` parameters are only required if the @@ -518,6 +523,9 @@ def __init__(self, super().__init__() def _create_connection_to_server(self): + con_timeout = self._connection_info['connection_timeout']\ + if self._connection_info['connection_timeout'] is not None\ + else self._connection_info['timeout'] object.__setattr__( self, '_handle', @@ -527,7 +535,7 @@ def _create_connection_to_server(self): self._connection_info['credentials'], self._connection_info['unix'], self._connection_info['tls'], - self._connection_info['timeout'], + con_timeout, self._connection_info['large_messages'] ) ) @@ -564,7 +572,8 @@ def _init(self, conn_gc_time: float = 0.0, reconnection_attempts: int = -1, reconnection_delay: float = 0.5, - reconnection_function: callable = reconnection_function + reconnection_function: callable = reconnection_function, + connection_timeout: Optional[float] = None, ): credentials = f'{normalize_to_str(username, "Username")}:' \ f'{normalize_to_str(password, "Password")}' @@ -585,7 +594,8 @@ def _init(self, 'conn_gc_time': conn_gc_time, 'reconnection_attempts': reconnection_attempts, 'reconnection_delay': reconnection_delay, - 'reconnection_function': reconnection_function + 'reconnection_function': reconnection_function, + 'connection_timeout': connection_timeout, }) if system == 'Windows' and unix: # nocov raise TypeError('Unix domain sockets cannot be used on Windows') @@ -600,13 +610,14 @@ def _init(self, object.__setattr__(self, '_handle', server_sock.fileno()) object.__setattr__(self, '_finalizer', lambda: server_sock.close()) else: + con_timeout = connection_timeout if connection_timeout is not None else timeout try: handle = _ipc.init_handle(host, port, credentials, unix, tls, - timeout, + con_timeout, large_messages) except BaseException as e: if isinstance(e, QError): @@ -1140,7 +1151,8 @@ def __init__(self, no_ctx: bool = False, reconnection_attempts: int = -1, reconnection_delay: float = 0.5, - reconnection_function: callable = reconnection_function + reconnection_function: callable = reconnection_function, + connection_timeout: Optional[float] = None, ): """Interface with a q process using the q IPC protocol. @@ -1181,6 +1193,9 @@ def __init__(self, `#!python reconnection_delay` on successive attempts to reconnect to the server. By default this is an exponential backoff where the `#!python reconnection_delay` is multiplied by two on each invocation. + connection_timeout: Timeout in seconds for connection to `q` server. If left as default + `None`, the socket will not timeout when connecting. + Note: The `#!python username` and `#!python password` parameters are not required. The `#!python username` and `#!python password` parameters are only required if the @@ -1247,7 +1262,9 @@ def __init__(self, no_ctx=no_ctx, reconnection_attempts=reconnection_attempts, reconnection_delay=reconnection_delay, - reconnection_function=reconnection_function + reconnection_function=reconnection_function, + connection_timeout=connection_timeout, + ) super().__init__() @@ -1453,7 +1470,9 @@ def __init__(self, no_ctx: bool = False, reconnection_attempts: int = -1, reconnection_delay: float = 0.5, - reconnection_function: callable = reconnection_function + reconnection_function: callable = reconnection_function, + connection_timeout: Optional[float] = None, + ): """Interface with a q process using the q IPC protocol. @@ -1499,6 +1518,9 @@ def __init__(self, `#!python reconnection_delay` on successive attempts to reconnect to the server. By default this is an exponential backoff where the `#!python reconnection_delay` is multiplied by two on each invocation + connection_timeout: Timeout in seconds for connection to `q` server. If left as default + `None`, the socket will not timeout when connecting. + Note: The `#!python username` and `#!python password` parameters are not required. The `#!python username` and `#!python password` parameters are only required if @@ -1593,6 +1615,7 @@ async def main(): 'reconnection_attempts':reconnection_attempts, 'reconnection_delay': reconnection_delay, 'reconnection_function': reconnection_function, + 'connection_timeout': connection_timeout, }) object.__setattr__(self, '_initialized', False) @@ -1613,6 +1636,7 @@ async def _async_init(self, reconnection_attempts: int = -1, reconnection_delay: float = 0.5, reconnection_function: callable = reconnection_function, + connection_timeout: float = 0.0, ): object.__setattr__(self, '_call_stack', []) self._init(host, @@ -1630,6 +1654,7 @@ async def _async_init(self, reconnection_attempts=reconnection_attempts, reconnection_delay=reconnection_delay, reconnection_function=reconnection_function, + connection_timeout=connection_timeout, ) object.__setattr__(self, '_loop', event_loop) con_info = object.__getattribute__(self, '_connection_info') @@ -1657,6 +1682,7 @@ async def _initobj(self): # nocov reconnection_attempts=self._stored_args['reconnection_attempts'], reconnection_delay=self._stored_args['reconnection_delay'], reconnection_function=self._stored_args['reconnection_function'], + connection_timeout=self._stored_args['connection_timeout'], ) return self @@ -2055,6 +2081,7 @@ def __init__(self, no_ctx: bool = False, as_server: bool = False, conn_gc_time: float = 0.0, + connection_timeout: Optional[float] = None, ): """Interface with a q process using the q IPC protocol. @@ -2093,6 +2120,9 @@ def __init__(self, going through the list of opened connections and closing any that the clients have closed. If not set the default of 0.0 will cause any old connections to never be closed unless `#!python self.clean_open_connections()` is manually called. + connection_timeout: Timeout in seconds for connection to `q` server. If left as default + `None`, the socket will not timeout when connecting. + Note: The `#!python username` and `#!python password` parameters are not required. The `#!python username` and `#!python password` parameters are only required if the q @@ -2137,7 +2167,7 @@ def __init__(self, ``` """ if timeout > 0.0: - warnings.warn('Timeout is not supported when using AsyncQConnection objects.') + warnings.warn('Timeout is not supported when using RawQConnection objects.') # TODO: Remove this once TLS support is fixed if tls: raise PyKXException('TLS is currently only supported for SyncQConnections') @@ -2164,6 +2194,7 @@ def __init__(self, 'no_ctx': True if as_server else no_ctx, 'as_server': as_server, 'conn_gc_time': conn_gc_time, + 'connection_timeout': connection_timeout, }) object.__setattr__(self, '_initialized', False) @@ -2182,6 +2213,7 @@ async def _async_init(self, no_ctx: bool = False, as_server: bool = False, conn_gc_time: float = 0.0, + connection_timeout: Optional[float] = None, ): object.__setattr__(self, '_call_stack', []) object.__setattr__(self, '_send_stack', []) @@ -2198,6 +2230,7 @@ async def _async_init(self, no_ctx=no_ctx, as_server=as_server, conn_gc_time=conn_gc_time, + connection_timeout=connection_timeout, ) object.__setattr__(self, '_loop', event_loop) con_info = object.__getattribute__(self, '_connection_info') @@ -2222,6 +2255,7 @@ async def _initobj(self): # nocov no_ctx=self._stored_args['no_ctx'], as_server=self._stored_args['as_server'], conn_gc_time=self._stored_args['conn_gc_time'], + connection_timeout=self._stored_args['connection_timeout'], ) return self @@ -2788,6 +2822,7 @@ def __init__(self, reconnection_attempts: int = -1, reconnection_delay: float = 0.5, reconnection_function: callable = reconnection_function, + connection_timeout: Optional[float] = None, ): """Interface with a q process using the q IPC protocol. @@ -2828,6 +2863,9 @@ def __init__(self, `#!python reconnection_delay` on successive attempts to reconnect to the server. By default this is an exponential backoff where the `#!python reconnection_delay` is multiplied by two on each invocation + connection_timeout: Timeout in seconds for connection to `q` server. If left as default + `None`, the socket will not timeout when connecting. + Note: The `#!python username` and `#!python password` parameters are not required. The `#!python username` and `#!python password` parameters are only required if @@ -2869,6 +2907,7 @@ def __init__(self, reconnection_attempts=reconnection_attempts, reconnection_delay=reconnection_delay, reconnection_function=reconnection_function, + connection_timeout=connection_timeout, ) super().__init__() diff --git a/src/pykx/pykxq_m.c b/src/pykx/pykxq_m.c index f3515e0..48de3bc 100644 --- a/src/pykx/pykxq_m.c +++ b/src/pykx/pykxq_m.c @@ -157,7 +157,7 @@ EXPORT K k_pykx_init(K k_q_lib_pat, K _pykx_threading) { return (K)0; } -void* thread_init(); +void* thread_init(void* args); EXPORT K k_init_python(K x, K y, K z) { pthread_mutex_init(&head_mutex, NULL); pthread_mutex_init(&cond_mutex, NULL); diff --git a/tests/test_db.py b/tests/test_db.py index 0c3b3ac..d18014e 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -294,6 +294,7 @@ def test_compress(kx): assert compress_info['zipLevel'].py() == 8 +@pytest.mark.order(20) def test_enumerate(kx): tab = kx.Table(data={ 'date': kx.q('2015.01.01 2015.01.01 2015.01.02 2015.01.02'), @@ -313,6 +314,7 @@ def test_enumerate(kx): assert type(kx.q.value(entab1['sym'])) == kx.SymbolVector # noqa: E721 +@pytest.mark.order(21) def test_partition_count(kx): db = kx.DB(path='db') fullview = db.partition_count() @@ -325,6 +327,7 @@ def test_partition_count(kx): assert all(cache == kx.q.Q.pv) +@pytest.mark.order(22) def test_subview(kx): db = kx.DB(path='db') db.subview([kx.q('2015.01.01')]) @@ -407,6 +410,74 @@ def test_load_failure(kx): os.remove('db/test.q') +@pytest.mark.order(23) +def test_sym_enum_create_1(kx): + shutil.rmtree('db') + db = kx.DB(path='db') + N = 333 + qtab = kx.Table(data={ + 'sym': kx.random.random(N, ['AAPL', 'GOOG', 'MSFT']), + 'price': kx.random.random(N, 10.0), + 'size': kx.random.random(N, 100) + }) + db.create(qtab, 'stocks', kx.q('2020.04m'), by_field='sym', sym_enum='junk') + test_path = db.path/"junk" + assert test_path.is_file() + + +@pytest.mark.order(24) +def test_sym_enum_load_1(kx): + db2 = kx.DB() + db2.load("db") + assert db2.tables == ['stocks'] + assert db2.stocks.size == kx.toq(1332) + + +@pytest.mark.order(25) +def test_sym_enum_create_2(kx): + db = kx.DB(path='db') + N = 333 + qtab = kx.Table(data={ + 'sym': kx.random.random(N, ['AAPL', 'GOOG', 'MSFT']), + 'price': kx.random.random(N, 10.0), + 'size': kx.random.random(N, 100) + }) + db.create(qtab, 'stocks', kx.q('2020.04m'), by_field='sym') + test_path = db.path/"sym" + assert test_path.is_file() + + +@pytest.mark.order(26) +def test_sym_enum_load_2(kx): + db2 = kx.DB() + db2.load('db') + assert db2.tables == ['stocks'] + assert db2.stocks.size == kx.toq(1332) + + +@pytest.mark.order(27) +def test_sym_enum_create_3(kx): + db = kx.DB(path='db') + N = 333 + qtab = kx.Table(data={ + 'sym': kx.random.random(N, ['AAPL', 'GOOG', 'MSFT']), + 'price': kx.random.random(N, 10.0), + 'size': kx.random.random(N, 100) + }) + db.create(qtab, 'stocks', kx.q('2020.04m'), by_field='sym', format="splayed", + sym_enum="test_junk") + test_path = db.path/"test_junk" + assert test_path.is_file() + + +@pytest.mark.order(28) +def test_sym_enum_load_3(kx): + db2 = kx.DB() + db2.load('db') + assert db2.tables == ['stocks'] + assert db2.stocks.size == kx.toq(1332) + + @pytest.mark.order(-1) def test_cleanup(kx): shutil.rmtree('db') diff --git a/tests/test_ipc.py b/tests/test_ipc.py index 73c0916..c61c8ef 100644 --- a/tests/test_ipc.py +++ b/tests/test_ipc.py @@ -1267,3 +1267,110 @@ async def test_async_reponse(kx, q_port): q('{t: .z.p;while[.z.p < t+00:00:02; neg[.z.w]99]}[]', reuse=False, async_response=True) with pytest.warns(UserWarning, match='Cannot use async_response=True without reuse=False.'): q('{t: .z.p;while[.z.p < t+00:00:02; neg[.z.w]99]}[]', wait=False, async_response=True) + + +def test_secure_connection_timeout(kx, q_port): + conn = kx.SecureQConnection(port=q_port, timeout=2.0, connection_timeout=1.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0) + and (info['connection_timeout']==1.0)) + + conn = conn= kx.SecureQConnection(port=q_port) + info = conn._connection_info + assert (info['port'] == q_port) + + conn = kx.SecureQConnection(port=q_port, timeout=2.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0)) + + conn = kx.SecureQConnection(port=q_port, connection_timeout=1.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['connection_timeout']==1.0)) + + conn = kx.SecureQConnection(port=q_port, timeout=2.0, connection_timeout=0.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0) + and (info['connection_timeout']==0.0)) + + conn = kx.SecureQConnection(port=q_port, timeout=2.0, connection_timeout=None) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0) + and (info['connection_timeout'] is None)) + + conn = kx.SecureQConnection(port=q_port, timeout=0.0, connection_timeout=None) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 0.0) + and (info['connection_timeout'] is None)) + + conn = kx.SecureQConnection(port=q_port, timeout=0.0, connection_timeout=0.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 0.0) + and (info['connection_timeout']==0.0)) + + +def test_connection_timeout(kx, q_port): + conn = kx.QConnection(port=q_port, timeout=2.0, connection_timeout=1.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0) + and (info['connection_timeout']==1.0)) + + conn = conn= kx.QConnection(port=q_port) + info = conn._connection_info + assert (info['port'] == q_port) + + conn = kx.QConnection(port=q_port, timeout=2.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0)) + + conn = kx.QConnection(port=q_port, connection_timeout=1.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['connection_timeout']==1.0)) + + conn = kx.QConnection(port=q_port, timeout=2.0, connection_timeout=0.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0) + and (info['connection_timeout']==0.0)) + + conn = kx.QConnection(port=q_port, timeout=2.0, connection_timeout=None) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 2.0) + and (info['connection_timeout'] is None)) + + conn = kx.QConnection(port=q_port, timeout=0.0, connection_timeout=None) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 0.0) + and (info['connection_timeout'] is None)) + + conn = kx.QConnection(port=q_port, timeout=0.0, connection_timeout=0.0) + info = conn._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 0.0) + and (info['connection_timeout']==0.0)) + + +@pytest.mark.asyncio +async def test_async_connection_timeout(kx, q_port): + async with kx.AsyncQConnection(port=q_port, connection_timeout=1.0) as q: + info = q._connection_info + assert ((info['port'] == q_port) and (info['connection_timeout'] == 1.0)) + + async with kx.AsyncQConnection(port=q_port) as q: + info = q._connection_info + assert (info['port'] == q_port) + + async with kx.AsyncQConnection(port=q_port, connection_timeout=0.0) as q: + info = q._connection_info + assert ((info['port'] == q_port) and (info['connection_timeout']==0.0)) + + async with kx.AsyncQConnection(port=q_port, connection_timeout=None) as q: + info = q._connection_info + assert ((info['port'] == q_port) and (info['connection_timeout'] is None)) + + async with kx.AsyncQConnection(port=q_port, timeout=0.0, connection_timeout=None) as q: + info = q._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 0.0) + and (info['connection_timeout'] is None)) + + async with kx.AsyncQConnection(port=q_port, timeout=0.0, connection_timeout=0.0) as q: + info = q._connection_info + assert ((info['port'] == q_port) and (info['timeout'] == 0.0) + and (info['connection_timeout']==0.0))