why can't connect to databases via airflow connection?

Question

After pushing my DAG I get this error

I am new to data engineering. I tried to solve this error in different ways at the expense of my knowledge, but nothing worked. I want to write a DAG that consists of two tasks, the first is to export data from one database table on one server as CSV files and import these CSV files into database tables on another server. The variable contains DAG configuration and SQL scripts for exporting and importing data.

Please tell me how can I solve this error?

I have this exporting code:

def export_csv():
    import json
    from airflow.models import Variable
    import pandas as pd
    instruction_data = json.loads(Variable.get('MAIN_SOURCE_DAMDI_INSTRUCTIONS'))
    requirement_data = instruction_data['requirements']
    lst = requirement_data['scripts']
    ms_hook = MsSqlHook(mssql_conn_id='OKTELL')
    connection = ms_hook.get_conn()
    cursor = connection.cursor()
    for i in lst:
        result = cursor.execute(i['export_script'])
        df = pd.DataFrame(result)
        df.to_csv(i['filename'], index=False, header=None, sep=',', encoding='utf-8')
    cursor.close()

And this is my task for exporting:

export_csv_func = PythonOperator(
    task_id='export_csv_func',
    python_callable=export_csv,
    mssql_conn_id='OKTELL'

P.S. I imported the libraries and airflow variables inside the function because before that there was a lot of load on the server and this method helped to reduce the load.

Answer 1

When using the PythonOperator you pass args to a callable via op_args and/or op_kwargs. In this case, if you wanted to pass the mssql_conn_id arg you can try:

export_csv_func = PythonOperator(
    task_id='export_csv_func',
    python_callable=export_csv,
    op_kwargs={'mssql_conn_id': 'OKTELL'},
)

Then you need to update the export_csv() function signature to accept this kwarg too.