Type signature for Patcher

kittens/transfer/algorithm.c

@@ -141,6 +141,7 @@ rolling_checksum_add_one_byte(rolling_checksum *self, uint8_t first_byte, uint8_
 
 // Python interface {{{
 
+// Patcher {{{
 typedef struct {
     PyObject_HEAD
     rolling_checksum rc;
@@ -155,8 +156,8 @@ static int
 Patcher_init(PyObject *s, PyObject *args, PyObject *kwds) {
     Patcher *self = (Patcher*)s;
     static char *kwlist[] = {"expected_input_size", NULL};
-    unsigned long long expected_input_size;
-    if (!PyArg_ParseTupleAndKeywords(args, kwds, "K", kwlist, &expected_input_size)) return -1;
+    unsigned long long expected_input_size = 0;
+    if (!PyArg_ParseTupleAndKeywords(args, kwds, "|K", kwlist, &expected_input_size)) return -1;
     self->block_size = default_block_size;
     if (expected_input_size > 0) {
         self->block_size = (size_t)round(sqrt((double)expected_input_size));
@@ -350,6 +351,7 @@ PyTypeObject Patcher_Type = {
     .tp_new = PyType_GenericNew,
     .tp_init = Patcher_init,
 };
+// }}} Patcher
 
 // Hasher {{{
 typedef struct {

kittens/transfer/rsync.pyi

@@ -1,13 +1,14 @@
-from typing import Union
+from typing import Callable, Union
 
-Buffer = Union[bytes, bytearray, memoryview]
+ReadOnlyBuffer = Union[bytes, bytearray, memoryview]
+WriteBuffer = Union[bytearray, memoryview]
 
 class RsyncError(Exception):
     pass
 
 class Hasher:
-    def __init__(self, which: str, data: Buffer = b''): ...
-    def update(self, data: Buffer) -> None: ...
+    def __init__(self, which: str, data: ReadOnlyBuffer = b''): ...
+    def update(self, data: ReadOnlyBuffer) -> None: ...
     def reset(self) -> None: ...
     def digest(self) -> bytes: ...
     def hexdigest(self) -> str: ...
@@ -18,3 +19,11 @@ class Hasher:
     def block_size(self) -> int: ...
     @property
     def name(self) -> str: ...
+
+
+class Patcher:
+
+    def __init__(self, expected_input_size: int = 0): ...
+    def signature_header(self, output: WriteBuffer) -> None: ...
+    def sign_block(self, block: ReadOnlyBuffer, output: WriteBuffer) -> None: ...
+    def apply_delta_data(self, data: ReadOnlyBuffer, read: Callable[[int, WriteBuffer], int], write: Callable[[ReadOnlyBuffer], None]) -> None: ...