Retrieve logs from SQL query in Airflow

[thegunner157]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from pendulum import datetime, duration, timezone
from airflow.utils.log.log_reader import TaskLogReader

def check_returncode(context, result):
    print("Result: ", result)
    print("Context: ", context)
    task_instance = context['task_instance']
    try_latest = task_instance.try_number
    task_log_reader = TaskLogReader()
    print(f"Try number: {try_latest}")
    logs = task_log_reader.read_log_chunks(task_instance, try_number=try_latest, metadata={"end_of_log": False})

    print(logs[0][0][0][1])

    print(logs)


with DAG(
        'Example-dag',
        default_args=default_args,
        schedule_interval=None,
        start_date=local_tz.convert(datetime(2023, 1, 16, 0, 0)),
        catchup=False,
        max_active_runs=1,
) as dag:
task = SQLExecuteQueryOperator(
        task_id='TASK_ID',
        sql="""
            Declare @execution_id bigint
            EXEC [SSISDB].[catalog].[create_execution]
                @folder_name = N'FOLDER',
                @project_name = N'PROJECT',
                @package_name = N'package.dtsx',
                @execution_id = @execution_id OUTPUT;
            Select @execution_id
            EXEC [SSISDB].[catalog].[start_execution]
                @execution_id;
        """,
        show_return_value_in_logs=True,
        email=None,
        post_execute=check_returncode,
    )

Description

Hi guys,
I want to retrieve logs from executed SQL stored procedure. Procedure does not return anything, but I want to get logs from it.
I tried to use SQLExecuteQueryOperator which calls SQL command and in post_execute function I tried to use TaskLogReader object and call read_log_chunks function with passed proper task_instance, try_number and metadata either as None or {end_of_log: True}.
However I constantly get error that I couldn’t read served logs and then below I get DetachedInstanceError from sqlalchemy which says “Parent instance is not bound to Session; lazy load operation “trigger” cannot proceed.”
Now the question is, can I somehow in post_execute function change task state to deferred or change the sql session from task instance level to enable logs? Or is there any solution that I can retrieve those logs from SQL?
I’m using Airflow in version 2.7.1.
Thanks guys in advance!

Operating System

Windows

Operating System Details

No response

SQLModel Version

I use qlalchemy

Python Version

3.9

Additional Context

No response

[Afstkla]

Hi, what's the relation between this question and SQLModel? This is a place to ask (and answer) questions about SQLModel. Seems like your question is related to airflow...

[thegunner157]

Yes, I'm sorry - wrong place