example.py

#!/usr/bin/env python3

import time

from .client import Client
from .protocol import EntryProtocol_pb2
from .protocol import CanalProtocol_pb2

client = Client()
client.connect(host='172.12.0.13')
client.check_valid()
client.subscribe()

while True:
    message = client.get(100)
    entries = message['entries']
    for entry in entries:
        entry_type = entry.entryType
        if entry_type in [EntryProtocol_pb2.EntryType.TRANSACTIONBEGIN, EntryProtocol_pb2.EntryType.TRANSACTIONEND]:
            continue
        row_change = EntryProtocol_pb2.RowChange()
        row_change.MergeFromString(entry.storeValue)
        event_type = row_change.eventType
        header = entry.header
        database = header.schemaName
        table = header.tableName
        event_type = header.eventType
        for row in row_change.rowDatas:
            format_data = dict()
            if event_type == EntryProtocol_pb2.EventType.DELETE:
                format_data = {
                    column.name: column.value
                    for column in row.beforeColumns
                }
            elif event_type == EntryProtocol_pb2.EventType.INSERT:
                format_data = {
                    column.name: column.value
                    for column in row.afterColumns
                }
            else:
                format_data['before'] = format_data['after'] = dict()
                for column in row.beforeColumns:
                    format_data['before'][column.name] = column.value
                for column in row.afterColumns:
                    format_data['after'][column.name] = column.value
            data = dict(
                db=database,
                table=table,
                event_type=event_type,
                data=format_data,
            )
            print(data)
    time.sleep(1)

client.disconnect()