diff --git a/pymysqlreplication/bitmap.py b/pymysqlreplication/bitmap.py index fd2096b6..54096dea 100644 --- a/pymysqlreplication/bitmap.py +++ b/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) diff --git a/pymysqlreplication/table.py b/pymysqlreplication/table.py index ed473cd6..3e34c169 100644 --- a/pymysqlreplication/table.py +++ b/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 diff --git a/pymysqlreplication/tests/test_data_objects.py b/pymysqlreplication/tests/test_data_objects.py index dc1281c3..fff03e80 100644 --- a/pymysqlreplication/tests/test_data_objects.py +++ b/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): diff --git a/pymysqlreplication/tests/test_data_type.py b/pymysqlreplication/tests/test_data_type.py index cd6464c5..7a8a917f 100644 --- a/pymysqlreplication/tests/test_data_type.py +++ b/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: