mysql-binlog-tutorial

read_event_query_1.py

import struct
import pendulum


class Header:
    def __init__(self):
        self.timestamp = 0
        self.event_type = None
        self.server_id = None
        self.event_size = 0
        self.log_pos = 0
        self.flag = 0

    def unpack(self, data):
        unpacked = struct.unpack('<IBIIIH', data)
        self.timestamp = unpacked[0]
        self.event_type = unpacked[1]
        self.server_id = unpacked[2]
        self.event_size = unpacked[3]
        self.log_pos = unpacked[4]
        self.flag = unpacked[5]

    def __repr__(self):
        fmt = pendulum.from_timestamp(self.timestamp).to_datetime_string()
        return f"timestamp:{self.timestamp} ({fmt})\n"\
          f"event_type:{self.event_type}\n"\
          f"server_id: {self.server_id}\n"\
          f"event_size: {self.event_size}\n"\
          f"next_log_pos: {self.log_pos}"
          
bin_log_file = "mysql-bin-changelog.000050.bin"
with open(bin_log_file, 'rb') as f:
  f.seek(462)
  header = Header()
  header.unpack(f.read(19))
  print(header)