From 8b5d58a0624b10d5bf1d8deba2c558b22343e14f Mon Sep 17 00:00:00 2001 From: yangbaechu Date: Wed, 30 Aug 2023 17:32:32 +0900 Subject: [PATCH 1/4] add typing in binlogfilereader.py --- pymysqlreplication/tests/binlogfilereader.py | 60 ++++++++++---------- 1 file changed, 30 insertions(+), 30 deletions(-) diff --git a/pymysqlreplication/tests/binlogfilereader.py b/pymysqlreplication/tests/binlogfilereader.py index 7075039e..75579c1d 100644 --- a/pymysqlreplication/tests/binlogfilereader.py +++ b/pymysqlreplication/tests/binlogfilereader.py @@ -6,22 +6,22 @@ from pymysqlreplication.event import QueryEvent from pymysqlreplication.event import RotateEvent from pymysqlreplication.event import XidEvent +from pymysqlreplication.event import BinLogEvent from pymysqlreplication.row_event import TableMapEvent from pymysqlreplication.row_event import WriteRowsEvent +from typing import Optional, List, Type, Union, Iterator, IO -class SimpleBinLogFileReader(object): - '''Read binlog files''' - +class SimpleBinLogFileReader: _expected_magic = b'\xfebin' - def __init__(self, file_path, only_events=None): - self._current_event = None - self._file = None - self._file_path = file_path - self._only_events = only_events - self._pos = None + def __init__(self, file_path: str, only_events: Optional[List[BinLogEvent]] = None) -> None: + self._current_event: Optional[SimpleBinLogEvent] = None + self._file: Optional[IO[bytes]] = None + self._file_path: str = file_path + self._only_events: Optional[List[BinLogEvent]] = only_events + self._pos: Optional[int] = None - def fetchone(self): + def fetchone(self) -> Optional["SimpleBinLogEvent"]: '''Fetch one record from the binlog file''' if self._pos is None or self._pos < 4: self._read_magic() @@ -33,12 +33,12 @@ def fetchone(self): if self._filter_events(event): return event - def truncatebinlog(self): + def truncatebinlog(self) -> None: '''Truncate the binlog file at the current event''' if self._current_event is not None: self._file.truncate(self._current_event.pos) - def _filter_events(self, event): + def _filter_events(self, event: BinLogEvent) -> bool: '''Return True if an event can be returned''' # It would be good if we could reuse the __event_map in # packet.BinLogPacketWrapper. @@ -52,14 +52,14 @@ def _filter_events(self, event): }.get(event.event_type) return event_type in self._only_events - def _open_file(self): + def _open_file(self) -> None: '''Open the file at ``self._file_path``''' if self._file is None: self._file = open(self._file_path, 'rb+') self._pos = self._file.tell() assert self._pos == 0 - def _read_event(self): + def _read_event(self) -> Optional["SimpleBinLogEvent"]: '''Read an event from the binlog file''' # Assuming a binlog version > 1 headerlength = 19 @@ -80,7 +80,7 @@ def _read_event(self): event.set_body(body) return event - def _read_magic(self): + def _read_magic(self) -> None: '''Read the first four *magic* bytes of the binlog file''' self._open_file() if self._pos == 0: @@ -92,10 +92,10 @@ def _read_magic(self): message = messagefmt.format(magic, self._expected_magic) raise BadMagicBytesError(message) - def __iter__(self): + def __iter__(self) -> Iterator[Optional["SimpleBinLogEvent"]]: return iter(self.fetchone, None) - def __repr__(self): + def __repr__(self) -> str: cls = self.__class__ mod = cls.__module__ name = cls.__name__ @@ -108,28 +108,28 @@ def __repr__(self): class SimpleBinLogEvent(object): '''An event from a binlog file''' - def __init__(self, header): + def __init__(self, header: bytes) -> None: '''Initialize the Event with the event header''' unpacked = struct.unpack(' None: '''Save the body bytes''' self.body = body - def set_pos(self, pos): + def set_pos(self, pos: int) -> None: '''Save the event position''' self.pos = pos - def __repr__(self): + def __repr__(self) -> str: cls = self.__class__ mod = cls.__module__ name = cls.__name__ @@ -146,4 +146,4 @@ class BadMagicBytesError(Exception): '''The binlog file magic bytes did not match the specification''' class EventSizeTooSmallError(Exception): - '''The event size was smaller than the length of the event header''' + '''The event size was smaller than the length of the event header''' \ No newline at end of file From 620036d04174a73a119c3abc1d47cd9b709576b5 Mon Sep 17 00:00:00 2001 From: yangbaechu Date: Wed, 30 Aug 2023 18:33:07 +0900 Subject: [PATCH 2/4] Fix typo --- pymysqlreplication/tests/binlogfilereader.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pymysqlreplication/tests/binlogfilereader.py b/pymysqlreplication/tests/binlogfilereader.py index 75579c1d..147d3a47 100644 --- a/pymysqlreplication/tests/binlogfilereader.py +++ b/pymysqlreplication/tests/binlogfilereader.py @@ -11,7 +11,7 @@ from pymysqlreplication.row_event import WriteRowsEvent from typing import Optional, List, Type, Union, Iterator, IO -class SimpleBinLogFileReader: +class SimpleBinLogFileReader(object): _expected_magic = b'\xfebin' def __init__(self, file_path: str, only_events: Optional[List[BinLogEvent]] = None) -> None: From 96168cd9cb6d5e59310de37a6a1b4ada175d2b69 Mon Sep 17 00:00:00 2001 From: yangbaechu Date: Wed, 30 Aug 2023 20:19:23 +0900 Subject: [PATCH 3/4] Fix annotation --- pymysqlreplication/tests/binlogfilereader.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pymysqlreplication/tests/binlogfilereader.py b/pymysqlreplication/tests/binlogfilereader.py index 147d3a47..68a06179 100644 --- a/pymysqlreplication/tests/binlogfilereader.py +++ b/pymysqlreplication/tests/binlogfilereader.py @@ -12,6 +12,7 @@ from typing import Optional, List, Type, Union, Iterator, IO class SimpleBinLogFileReader(object): + '''Read binlog files''' _expected_magic = b'\xfebin' def __init__(self, file_path: str, only_events: Optional[List[BinLogEvent]] = None) -> None: From 97b6c2fb5a115c2b546dc6bdb6dea9cde7fd69f1 Mon Sep 17 00:00:00 2001 From: yangbaechu Date: Wed, 30 Aug 2023 20:23:09 +0900 Subject: [PATCH 4/4] Fix typo --- pymysqlreplication/tests/binlogfilereader.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pymysqlreplication/tests/binlogfilereader.py b/pymysqlreplication/tests/binlogfilereader.py index 68a06179..3b197a84 100644 --- a/pymysqlreplication/tests/binlogfilereader.py +++ b/pymysqlreplication/tests/binlogfilereader.py @@ -13,6 +13,7 @@ class SimpleBinLogFileReader(object): '''Read binlog files''' + _expected_magic = b'\xfebin' def __init__(self, file_path: str, only_events: Optional[List[BinLogEvent]] = None) -> None: