add ty type checker

.github/workflows/check.yml

@@ -14,17 +14,23 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-python@v3
       with:
         python-version: ${{ matrix.python-version }}
+    - run: python -m venv .venv
+    - run: |
+        . .venv/bin/activate
+        echo "$VIRTUAL_ENV/bin" >> $GITHUB_PATH
+        env
     - run: python -m pip install --upgrade pip
     - run: python -m pip install "ruff<1" "mypy<2" pytest
     - run: pip install -r requirements.txt
     - run: ruff check .
     - run: ruff format --check .
+    - run: ty check
     - run: mypy --strict .
     - run: pytest .

README.md

@@ -2,7 +2,7 @@
 [![FOSSA Status](https://app.fossa.com/api/projects/git%2Bgithub.com%2FTeaEngineering%2Faiokdb.svg?type=shield)](https://app.fossa.com/projects/git%2Bgithub.com%2FTeaEngineering%2Faiokdb?ref=badge_shield)
 
 # aiokdb
-Python asyncio connector to KDB. Pure python, so does not depend on the `k.h` bindings or kdb shared objects, or numpy/pandas. Fully type hinted to comply with `PEP-561`. No non-core dependencies, and tested on Python 3.8 - 3.12.
+Python asyncio connector to KDB. Pure python, so does not depend on the `k.h` bindings or kdb shared objects, or numpy/pandas. Fully type hinted to comply with `PEP-561`. No non-core dependencies, and tested on Python 3.10 - 3.13. Older versions supported python versions 3.8 and 3.9
 
 ## Peer review & motivation
 

aiokdb/__init__.py

@@ -258,7 +258,7 @@ def kE(self) -> "MutableSequence[float]":
     def kF(self) -> "MutableSequence[float]":
         raise self._te()
 
-    def kC(self) -> array.array:  # type: ignore[type-arg]
+    def kC(self) -> "array.array[str]":
         raise self._te()
 
     def kS(self) -> "MutableSequence[str]":
@@ -824,7 +824,7 @@ def _databytes(self) -> bytes:
     def __repr__(self) -> str:
         return f"cv({repr(self.aS())})"
 
-    def kC(self) -> array.array:  # type: ignore[type-arg]
+    def kC(self) -> "array.array[str]":
         return self._c
 
     def aS(self) -> str:

aiokdb/adapter.py

@@ -17,37 +17,39 @@ def __init__(self, data: "MutableSequence[int]"):
         self.data = data
 
     @overload
-    def __getitem__(self, i: int) -> bool: ...
+    def __getitem__(self, index: int) -> bool: ...
     @overload
-    def __getitem__(self, s: slice) -> "MutableSequence[bool]": ...
-    def __getitem__(self, i: Union[int, slice]) -> Union[bool, "MutableSequence[bool]"]:
-        if isinstance(i, slice):
-            return self.__class__(self.data[i])
+    def __getitem__(self, index: slice) -> "MutableSequence[bool]": ...
+    def __getitem__(
+        self, index: Union[int, slice]
+    ) -> Union[bool, "MutableSequence[bool]"]:
+        if isinstance(index, slice):
+            return self.__class__(self.data[index])
         else:
-            return {0: False, 1: True}[self.data[i]]
+            return {0: False, 1: True}[self.data[index]]
 
     def __len__(self) -> int:
         return len(self.data)
 
     @overload
-    def __setitem__(self, index: int, item: bool) -> None: ...
+    def __setitem__(self, index: int, value: bool) -> None: ...
     @overload
-    def __setitem__(self, index: slice, item: Iterable[bool]) -> None: ...
+    def __setitem__(self, index: slice, value: Iterable[bool]) -> None: ...
     def __setitem__(
-        self, index: Union[int, slice], item: Union[bool, Iterable[bool]]
+        self, index: Union[int, slice], value: Union[bool, Iterable[bool]]
     ) -> None:
-        if isinstance(index, slice) and isinstance(item, Iterable):
-            self.data[index] = array.array("B", [{True: 1, False: 0}[i] for i in item])
-        elif isinstance(index, int) and isinstance(item, bool):
-            self.data[index] = {True: 1, False: 0}[item]
+        if isinstance(index, slice) and isinstance(value, Iterable):
+            self.data[index] = array.array("B", [{True: 1, False: 0}[v] for v in value])
+        elif isinstance(index, int) and isinstance(value, bool):
+            self.data[index] = {True: 1, False: 0}[value]
         else:
             raise TypeError()
 
-    def insert(self, index: int, item: bool) -> None:
-        self.data.insert(index, {True: 1, False: 0}[item])
+    def insert(self, index: int, value: bool) -> None:
+        self.data.insert(index, {True: 1, False: 0}[value])
 
-    def __delitem__(self, item: Union[int, slice]) -> None:
-        del self.data[item]
+    def __delitem__(self, index: Union[int, slice]) -> None:
+        del self.data[index]
 
     def __eq__(self, other: Any) -> bool:
         if isinstance(other, Sequence):
@@ -64,37 +66,39 @@ def __init__(self, data: "MutableSequence[int]", context: KContext):
         self.context = context
 
     @overload
-    def __getitem__(self, i: int) -> str: ...
+    def __getitem__(self, index: int) -> str: ...
     @overload
-    def __getitem__(self, s: slice) -> "MutableSequence[str]": ...
-    def __getitem__(self, i: Union[int, slice]) -> Union[str, "MutableSequence[str]"]:
-        if isinstance(i, slice):
-            return self.__class__(self.data[i], self.context)
+    def __getitem__(self, index: slice) -> "MutableSequence[str]": ...
+    def __getitem__(
+        self, index: Union[int, slice]
+    ) -> Union[str, "MutableSequence[str]"]:
+        if isinstance(index, slice):
+            return self.__class__(self.data[index], self.context)
         else:
-            return self.context.lookup_str(self.data[i])
+            return self.context.lookup_str(self.data[index])
 
     def __len__(self) -> int:
         return len(self.data)
 
     @overload
-    def __setitem__(self, index: int, item: str) -> None: ...
+    def __setitem__(self, index: int, value: str) -> None: ...
     @overload
-    def __setitem__(self, index: slice, item: Iterable[str]) -> None: ...
+    def __setitem__(self, index: slice, value: Iterable[str]) -> None: ...
     def __setitem__(
-        self, index: Union[int, slice], item: Union[str, Iterable[str]]
+        self, index: Union[int, slice], value: Union[str, Iterable[str]]
     ) -> None:
-        if isinstance(index, slice) and isinstance(item, Iterable):
-            self.data[index] = array.array("l", [self.context.ss(i) for i in item])
-        elif isinstance(index, int) and isinstance(item, str):
-            self.data[index] = self.context.ss(item)
+        if isinstance(index, slice) and isinstance(value, Iterable):
+            self.data[index] = array.array("l", [self.context.ss(v) for v in value])
+        elif isinstance(index, int) and isinstance(value, str):
+            self.data[index] = self.context.ss(value)
         else:
             raise TypeError()
 
-    def insert(self, index: int, item: str) -> None:
-        self.data.insert(index, self.context.ss(item))
+    def insert(self, index: int, value: str) -> None:
+        self.data.insert(index, self.context.ss(value))
 
-    def __delitem__(self, item: Union[int, slice]) -> None:
-        del self.data[item]
+    def __delitem__(self, index: Union[int, slice]) -> None:
+        del self.data[index]
 
     def __eq__(self, other: Any) -> bool:
         if isinstance(other, Sequence):

aiokdb/format.py

@@ -180,7 +180,7 @@ def _fmt_atom_p(self, j: int) -> str:
         nanos = j % 1000
         micros = j // 1000
         origin = int(datetime(2000, 1, 1, tzinfo=timezone.utc).timestamp())
-        dt = datetime.utcfromtimestamp(origin + micros / 1000000.0)
+        dt = datetime.fromtimestamp(origin + micros / 1000000.0, timezone.utc)
         return dt.strftime("%Y.%m.%dD%H:%M:%S:%f") + f"{nanos:03}"
 
     def _fmt_atom_n(self, j: int) -> str:
@@ -206,14 +206,14 @@ def _fmt_atom_d(self, d: int) -> str:
         if d == Nulls.i:
             return "0Nd"
         origin = int(datetime(2000, 1, 1, tzinfo=timezone.utc).timestamp())
-        dt = datetime.utcfromtimestamp(origin + d)
+        dt = datetime.fromtimestamp(origin + d, tz=timezone.utc)
         return dt.strftime("%Y.%m.%d")
 
     def _fmt_atom_z(self, d: float) -> str:
         if d == Nulls.f:
             return "0Nz"
         origin = int(datetime(2000, 1, 1, tzinfo=timezone.utc).timestamp())
-        dt = datetime.utcfromtimestamp(origin + d)
+        dt = datetime.fromtimestamp(origin + d, tz=timezone.utc)
         return dt.strftime("%Y.%m.%dT%H:%M:%S:%f")
 
     def _fmt_atom_u(self, u: int) -> str:
@@ -331,25 +331,23 @@ def _fmt_inline(self, obj: KObj) -> str:
         raise ValueError(f"No inline formatter for {obj} with type {obj._tn()}")
 
 
-def identity(x: str) -> str:
-    return x
-
-
 class HtmlFormatter(AsciiFormatter):
     def __init__(
         self,
         table_class: Optional[str] = None,
         indent: int = 2,
         width: int = 200,
         height: int = 10,
-        markup: Callable[[str], str] = identity,
-        escape: Callable[[str], str] = escape,
     ):
         super().__init__(width, height)
         self.tc = f' class="{table_class}"' if table_class else ""
         self.indent = indent
-        self.markup = markup
-        self.escape = escape
+
+    def html_markup(self, text: str) -> str:
+        return text
+
+    def html_escape(self, text: str) -> str:
+        return escape(text)
 
     def format(self, obj: KObj) -> str:
         if obj.t == TypeEnum.XT:
@@ -358,7 +356,7 @@ def format(self, obj: KObj) -> str:
             return self._fmt_keyed_table(obj)
         elif obj.t == TypeEnum.XD:
             return self._fmt_dict(obj)
-        return self.escape(self._fmt_inline(obj))
+        return self.html_escape(self._fmt_inline(obj))
 
     def _fmt_unkeyed_table(self, obj: KObj) -> str:
         rowcount = self._table_conforms(obj)
@@ -402,10 +400,10 @@ def _fmt_unkeyed_table(self, obj: KObj) -> str:
                 "</table>",
             ]
         )
-        return self.markup(rowHtml)
+        return self.html_markup(rowHtml)
 
     def _html_cell(self, obj: KObj, col: int, index: Optional[int]) -> str:
-        return self.escape(self._str_cell(obj, col, index))
+        return self.html_escape(self._str_cell(obj, col, index))
 
     def get_table_cell_formatter_for(
         self, kob: KObj, isKey: bool, i: int, colName: str
@@ -428,7 +426,10 @@ def _fmt_keyed_table(self, obj: KObj) -> str:
             (obj, True, i, s) for i, s in enumerate(ktk.kS())
         ] + [(obj, False, i, s) for i, s in enumerate(ktv.kS())]
 
-        table_col_formatters = [self.get_table_cell_formatter_for(*x) for x in colMeta]
+        table_col_formatters = [
+            self.get_table_cell_formatter_for(kob, isKey, i, colName)
+            for kob, isKey, i, colName in colMeta
+        ]
         rowSample: List[List[Tuple[str, bool]]] = []
         for r in rows:
             cs = []
@@ -460,18 +461,18 @@ def _fmt_keyed_table(self, obj: KObj) -> str:
                 "</table>",
             ]
         )
-        return self.markup(rowHtml)
+        return self.html_markup(rowHtml)
 
     def _fmt_dict(self, obj: KObj) -> str:
         rows = list(self._select_rows(len(obj)))
         ks, vs = [], []
         for r in rows:
-            k = self.escape(self._str_cell(obj.kkey(), 0, r))
-            v = self.escape(self._str_cell(obj.kvalue(), 0, r))
+            k = self.html_escape(self._str_cell(obj.kkey(), 0, r))
+            v = self.html_escape(self._str_cell(obj.kvalue(), 0, r))
             ks.append(k)
             vs.append(v)
 
-        return self.markup(
+        return self.html_markup(
             "".join(
                 [
                     "<dl>\n",

aiokdb/server.py

@@ -47,7 +47,7 @@ async def _read(self) -> Tuple[MessageType, KObj]:
         if len(payload) < 1000 and logging.getLogger().isEnabledFor(logging.DEBUG):
             logger.debug(f"> recv buffer={msgh + payload!r}")
         k = d9(msgh + payload)
-        return msgtype, k
+        return MessageType(msgtype), k
 
     async def read(self) -> Tuple[MessageType, KObj]:
         msgtype, k = await self._read()
@@ -281,7 +281,7 @@ async def reader_to_context_task(
             else:
                 raise Exception(f"{q_writer.qid} Unexpected incoming message type")
 
-    except asyncio.exceptions.IncompleteReadError:
+    except asyncio.IncompleteReadError:
         # reader is closed, we have nothing more to send
         q_writer.close()
 
@@ -306,7 +306,7 @@ async def handle_connection(
         await reader_to_context_task(q_writer, q_reader, context)
     except asyncio.TimeoutError:
         logging.info(f"{qid} closed - login timeout")
-    except asyncio.exceptions.IncompleteReadError:
+    except asyncio.IncompleteReadError:
         # When kdb process timeout via .timer.timeoutSyncCall
         logging.log(disconnect_log_level, f"{qid} connection reached end of stream")
     except BrokenPipeError:

check.sh

@@ -1,5 +1,6 @@
 set -e
 ruff check .
 ruff format .
+ty check
 mypy --strict .
 pytest .

pyproject.toml

@@ -12,20 +12,22 @@ markers = [
     "asyncio",
 ]
 
+[tool.ty.rules]
+redundant-cast = "ignore"
+
 [project]
 name = "aiokdb"
 authors = [
   { name="Chris Shucksmith", email="chris@shucksmith.co.uk" },
 ]
 description = "Pure Python asyncio connector to KDB"
 readme = "README.md"
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 classifiers = [
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "License :: OSI Approved :: MIT License",
     "Operating System :: OS Independent",
 ]

requirements.minimal.txt

@@ -0,0 +1,5 @@
+prompt-toolkit
+pytest
+pytest-asyncio
+ty
+ruff

requirements.txt

@@ -1,3 +1,9 @@
-pytest
-pytest-asyncio
+iniconfig==2.1.0
+packaging==25.0
 prompt-toolkit==3.0.43
+Pygments==2.19.2
+pytest==9.0.2
+pytest-asyncio==1.3.0
+ruff==0.12.10
+ty==0.0.18
+wcwidth==0.2.13

test/test_format.py

@@ -139,8 +139,8 @@ def get_table_cell_formatter_for(
             return super().get_table_cell_formatter_for(kob, isKey, i, colName)
 
         def bold_html_cell(self, obj: KObj, col: int, index: Optional[int]) -> str:
-            return self.markup(
-                "<b>" + self.escape(self._str_cell(obj, col, index)) + "</b>"
+            return self.html_markup(
+                "<b>" + self.html_escape(self._str_cell(obj, col, index)) + "</b>"
             )
 
     b = SillyHtmlFormatter()