typing - 'test_data_type.py', 'table.py', 'test_data_objects.py', 'bitmap.py'

pymysqlreplication/bitmap.py

@@ -1,6 +1,8 @@
+from typing import List
+
 # -*- coding: utf-8 -*-
 
-bitCountInByte = [
+bitCountInByte: List[int] = [
     0, 1, 1, 2, 1, 2, 2, 3, 1, 2, 2, 3, 2, 3, 3, 4,
     1, 2, 2, 3, 2, 3, 3, 4, 2, 3, 3, 4, 3, 4, 4, 5,
     1, 2, 2, 3, 2, 3, 3, 4, 2, 3, 3, 4, 3, 4, 4, 5,
@@ -19,18 +21,21 @@
     4, 5, 5, 6, 5, 6, 6, 7, 5, 6, 6, 7, 6, 7, 7, 8,
 ]
 
+
 # Calculate total bit counts in a bitmap
-def BitCount(bitmap):
+def BitCount(bitmap: bytes) -> int:
     n = 0
     for i in range(0, len(bitmap)):
         bit = bitmap[i]
         if type(bit) is str:
             bit = ord(bit)
         n += bitCountInByte[bit]
+
     return n
 
+
 # Get the bit set at offset position in bitmap
-def BitGet(bitmap, position):
+def BitGet(bitmap: bytes, position: int) -> int:
     bit = bitmap[int(position / 8)]
     if type(bit) is str:
         bit = ord(bit)

pymysqlreplication/table.py

@@ -1,8 +1,13 @@
 # -*- coding: utf-8 -*-
+from typing import List
+from typing import Dict
+from typing import Optional
+from pymysqlreplication.column import Column
 
 
 class Table(object):
-    def __init__(self, column_schemas, table_id, schema, table, columns, primary_key=None):
+    def __init__(self, column_schemas: List[Dict[str, str]], table_id: int, schema: str, table: str,
+                 columns: List[Column], primary_key: Optional[List[str]] = None):
         if primary_key is None:
             primary_key = [c.data["name"] for c in columns if c.data["is_primary"]]
             if len(primary_key) == 0:
@@ -25,11 +30,11 @@ def __init__(self, column_schemas, table_id, schema, table, columns, primary_key
     def data(self):
         return dict((k, v) for (k, v) in self.__dict__.items() if not k.startswith('_'))
 
-    def __eq__(self, other):
+    def __eq__(self, other: 'Table') -> bool:
         return self.data == other.data
 
-    def __ne__(self, other):
+    def __ne__(self, other: 'Table') -> bool:
         return not self.__eq__(other)
 
-    def serializable_data(self):
+    def serializable_data(self) -> 'Table':
         return self.data

pymysqlreplication/tests/test_data_objects.py

@@ -9,12 +9,13 @@
 from pymysqlreplication.event import GtidEvent
 
 from pymysqlreplication.tests import base
+from typing import List
 
 __all__ = ["TestDataObjects"]
 
 
 class TestDataObjects(base.PyMySQLReplicationTestCase):
-    def ignoredEvents(self):
+    def ignoredEvents(self) -> List[GtidEvent]:
         return [GtidEvent]
 
     def test_column_is_primary(self):

pymysqlreplication/tests/test_data_type.py

@@ -34,7 +34,7 @@ class TestDataType(base.PyMySQLReplicationTestCase):
     def ignoredEvents(self):
         return [GtidEvent]
 
-    def create_and_insert_value(self, create_query, insert_query):
+    def create_and_insert_value(self, create_query: str, insert_query: str):
         self.execute(create_query)
         self.execute(insert_query)
         self.execute("COMMIT")
@@ -57,7 +57,7 @@ def create_and_insert_value(self, create_query, insert_query):
         self.assertIsInstance(event, WriteRowsEvent)
         return event
 
-    def create_table(self, create_query):
+    def create_table(self, create_query: str):
         """Create table 
 
         Create table in db and return query event.
@@ -77,7 +77,7 @@ def create_table(self, create_query):
 
         return event
 
-    def create_and_get_tablemap_event(self, bit):
+    def create_and_get_tablemap_event(self):
         """Create table and return tablemap event
 
         Returns: