纯Python asyncio连接器到KDB
项目描述
aiokdb
Python asyncio连接器到KDB。纯Python，因此不依赖于k.h绑定或kdb共享对象，也不依赖于numpy/pandas。完全类型提示，符合PEP-561。没有非核心依赖，并在Python 3.8 - 3.12上进行了测试。
同行评审和动机
qPython 是用于此任务的广泛使用的库，它将对象映射到Pandas Dataframes，这可能更适合大多数应用程序。
此库采用不同的方法，旨在使用KDB C库函数进行复制，即100%明确关于KDB类型。它基于公开文档的 序列化示例 和 kdb+ C API 页面构建。用户可能还需要熟悉 k.h。
一个简单的示例，使用阻塞套接字
# run ./q -p 12345 &
from aiokdb.socket import khpu
h = khpu("localhost", 12345, "kdb:pass")
# if remote returns Exception, it is raised here, unless khpu(..., raise_krr=False)
result = h.k("2.0+3.0")
assert result.aF() == 5.0
result.aJ() # raises ValueError: wrong type KF (-9) for aJ
结果对象是一个类似K的Python对象（一个KObj），拥有可用作result.type的常用有符号整型。原始类型访问器以a为前缀，并在运行时检查访问器是否适合存储的类型（例如.aI()、.aJ()、.aH()、.aF()等）。原子无论类型如何都将值存储到bytes对象中，并在需要时进行编码/解码。可以使用（.i(3)、.j(12)、.ss("hello")）设置原子值。
数组通过使用Python的本地数组模块来实现，以实现高效的数组类型。《可变序列》数组使用通常的数组访问函数返回，如.kI()、.kB()、.kS()等。
序列化由b9函数处理，该函数将KObj编码为Python bytes，而d9函数则接受bytes并返回KObj。
- 原子通过
ka、kb、ku、kg、kh、ki、kj、ke、kf、kc、ks、kt、kd、kz、ktj创建。 - 使用
ktn和knk的列表 - 使用
xd的字典和使用xt的表。
Python管理垃圾收集，因此不存在引用计数原始类型，即不存在k.r和函数r1、r0和m9、setm。
Asyncio
使用asyncio实现kdb客户端和服务器协议，并且可以相互测试。例如，运行python -m aiokdb.server然后运行python -m aiokdb.client将使用KDB IPC连接在一起。然而，由于没有解释器（默认服务器不处理任何命令），服务器将返回对所有查询的nyi错误。要为您的应用程序实现部分协议，请从aiokdb.server.ServerContext中派生，并实现on_sync_request()、on_async_message()，也许还有check_login()。
命令行界面
该软件包内置了对远程KDB实例的命令行客户端支持，可用于连接（使用python asyncio和prompt_toolkit进行行编辑和历史记录）。
$ pip install aiokdb prompt_toolkit
$ ./q -p 12345 &
$ python -m aiokdb.cli --host localhost --port 12345
(eval) > ([s:7 6 0Nj]x:3?0Ng;y:2)
s| x y
-|---------------------------------------
7| 409031f3-b19c-6770-ee84-6e9369c98697 2
6| 52cb20d9-f12c-9963-2829-3c64d8d8cb14 2
| cddeceef-9ee9-3847-9172-3e3d7ab39b26 2
(eval) > 4 5 6!(`abc`def;til 300;(3 4!`a`b))
4| abc def
5| 0 1 2 ... 297 298 299
6| KDict
(eval) > [ctrl-D]
$
上面的文本格式由aiokdb.format.ASCIIFormatter控制，该格式在KObj内部查找以将XD、SD、XT类型渲染为包含原子和向量值的表格形式。嵌套的复杂类型，例如字典或表，渲染为KDict或KFlip常量。
QDB文件
可以使用kfromfile读取使用set编写的普通.qdb文件，可以使用ktofile写入。
>>> from aiokdb.files import kfromfile, ktofile
>>> k = kfromfile('test_qdb0/test.qdb')
>>> k
<aiokdb.KObjArray object at 0x7559136d8230>
>>> from aiokdb.format import AsciiFormatter
>>> fmt = AsciiFormatter()
>>> print(fmt.format(k))
[5, hello]
通常k是一个字典，表示包含其他对象的KDB命名空间。
测试
该库具有广泛的测试覆盖，但是某些（不为人知的）KObj的反序列化可能尚未完全支持。欢迎提交PR。所有测试都是纯Python编写的，除了test/test_rpc.py中的测试，该测试将使用真实的KDB服务器进行测试，如果设置了KDB_PYTEST_SERVICE环境变量（形式为kdb://user:password@hostname:port），否则该测试将被跳过。
- 使用
ruff check .进行格式化 - 使用
ruff format .进行格式化 - 使用
mypy --strict .检查类型注释 - 在根目录中运行
pytest .
许可证
下载文件
下载适用于您平台的文件。如果您不确定选择哪个，请了解有关安装包的更多信息。
源代码分发
构建分发
aiokdb-0.1.14.tar.gz的哈希值
| 算法 | 哈希摘要 | |
|---|---|---|
| SHA256 | 51ef028fed5eb28d0f3d32199357189c45f549c85d48f554171f9308d312b846 |
|
| MD5 | a69fd228a79c70aae20f2c4745569e15 |
|
| BLAKE2b-256 | 2c857b6a82b18949fc6b5b31897556c08426965659107e4314176fae723ea6c1 |
aiokdb-0.1.14-py3-none-any.whl的哈希值
| 算法 | 哈希摘要 | |
|---|---|---|
| SHA256 | e72fffe7c086bb1d1c50b3355753b54f97905ebcb6e8b79b46cd1799acb56456 |
|
| MD5 | 6feccfe0a0eabcbdd7a0b3c50be10512 |
|
| BLAKE2b-256 | 390ec4e8811754f98ed8724808d1d60dbda975fb83b9352ea0825207997673f3 |