Apache Airflow Rows Affected Catcher

airflow_rows_affected.py

import logging
from logging import Handler, LogRecord

from airflow.operators.python import get_current_context


class RowsAffectedHandler(Handler):
    def emit(self, record: LogRecord) -> None:
        msg = self.format(record)
        context = get_current_context()
        result = re.search(r'Rows affected: (?P<rowscount>\d+)$', msg)
        if result:
            rows = result.groupdict().get('rowscount')
            context['ti'].xcom_push(value=f'{rows}', key='rows_affected')


handler = RowsAffectedHandler()
handler.addFilter(lambda record: record.getMessage().find('Rows affected') != -1)
logging.getLogger('airflow').addHandler(handler)

# ниже код вашего DAG