Clearing xcom data after 30 days

Airflow
1

I am trying to clear xcom values based on the execution_date older than 30 days, but the dag is throwing error as execution date is not defined . Please help to understand the issue here (airflow 2.4.3)
err : Xcom not defiend
code is follows

from airflow.models import DAG
from airflow.utils.db import provide_session
from airflow.models import XCom
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago
from datetime import datetime, timedelta,timezone
from sqlalchemy import func

with DAG(dag_id="cleanup_xcom_demo", schedule_interval=None, start_date=days_ago(2)) as dag:
    # cleanup_xcom
    @provide_session
    def cleanup_xcom(session=None, **context):
        dag = context["dag"]
        dag_id = dag._dag_id 
        # It will delete all xcom of the dag_id
        ts_limit = datetime.now(timezone.utc) - timedelta(days=30)
        print(f"printing session {session}")
        session.query(XCom).filter(Xcom.execution_date <= ts_limit ).delete()
   
    clean_xcom = PythonOperator(
        task_id="clean_xcom",
        python_callable = cleanup_xcom,
        provide_context=True, 
        # dag=dag
    )
    
    start  = DummyOperator(task_id="start")
    end = DummyOperator(task_id="end", trigger_rule="none_failed")
    
    start >> clean_xcom >> end
2

Hello @datagirl

There is a typo in your code. Instead of XCom you have Xcom. Also, I would recommend printing the count of rows to be deleted using count() before issuing the actual delete command.

Another cleaner way to write this is:

from airflow import settings
from airflow.utils.dates import days_ago
from airflow.models import XCom
from airflow.decorators import task

from datetime import datetime, timezone

with DAG(dag_id="cleanup_xcom_demo", schedule_interval=None, start_date=days_ago(2)) as dag:
    
    @task
    def cleanup_xcom():
        with settings.Session() as session:
            ts_limit = datetime.now(timezone.utc) - timedelta(days=30)
            print(ts_limit)
            cnt = session.query(XCom).filter(XCom.execution_date <= ts_limit).count()
            print(cnt)
    
    cleanup_xcom()

Also, please note DummyOperator has been deprecated. You can use EmptyOperator instead.

Thanks
Manmeet