Update airflow_db_cleanup.py to also clean sessions table (#10775)

* Update airflow_db_cleanup.py to also clean sessions table

* Update airflow_db_cleanup.py

Author: apilaskowski

composer/workflows/airflow_db_cleanup.py

@@ -68,7 +68,7 @@
 from airflow.version import version as airflow_version
 
 import dateutil.parser
-from sqlalchemy import and_, func
+from sqlalchemy import and_, func, text
 from sqlalchemy.exc import ProgrammingError
 from sqlalchemy.orm import load_only
 
@@ -461,16 +461,42 @@ def cleanup_function(**context):
         session.close()
 
 
+def cleanup_sessions():
+    session = settings.Session()
+
+    try:
+        logging.info("Deleting sessions...")
+        before = len(session.execute(text("SELECT * FROM session WHERE expiry > now()::timestamp(0);")).mappings().all())
+        session.execute(text("DELETE FROM session WHERE expiry > now()::timestamp(0);"))
+        after = len(session.execute(text("SELECT * FROM session WHERE expiry > now()::timestamp(0);")).mappings().all())
+        logging.info("Deleted {} expired sessions.".format(before-after))
+    except Exception as e:
+        logging.error(e)
+
+    session.commit()
+    session.close()
+
+
 def analyze_db():
     session = settings.Session()
     session.execute("ANALYZE")
     session.commit()
+    session.close()
 
 
 analyze_op = PythonOperator(
     task_id="analyze_query", python_callable=analyze_db, provide_context=True, dag=dag
 )
 
+cleanup_session_op = PythonOperator(
+    task_id="cleanup_sessions",
+    python_callable=cleanup_sessions,
+    provide_context=True,
+    dag=dag
+)
+
+cleanup_session_op.set_downstream(analyze_op)
+
 for db_object in DATABASE_OBJECTS:
     cleanup_op = PythonOperator(
         task_id="cleanup_" + str(db_object["airflow_db_model"].__name__),