def clean_dags_logs():
    hook = PostgresHook(postgres_conn_id='airflow_postgres')
    dag_files = Path(Path(__file__).parent).glob('*.py')
    dags = ['{p.parent.name}_{p.stem}'.format(p=p) for p in dag_files]
    dags = [d for d in dags if (d != DAG_ID)]
    execution_date = (datetime.date(datetime.now()) - timedelta(days=DAYS_TO_RETAIN))
    p = Path(LOG_DIR)
    for d in dags:
        logger.info('Cleaning up meta tables for {}'.format(d))
        for t in TABLES:
            sql = "delete from {} where dag_id='{}' and execution_date < '{}'".format(t, d, execution_date)
            hook.run(sql, True)
        logger.info('Cleaning up log folder for {}'.format(d))
        for path in list(p.glob('{}/*/*'.format(d))):
            log_date = str(path).split('/')[(- 1)]
            log_date = log_date.split('T')[0]
            log_date = datetime.date(datetime.strptime(log_date, '%Y-%m-%d'))
            if (log_date < execution_date):
                logger.info('Deleting dir {}'.format(str(path.absolute())))
                shutil.rmtree(str(path.absolute()))
def clean_dags_logs():
    hook = PostgresHook(postgres_conn_id='airflow_postgres')
    dag_files = Path(Path(__file__).parent).glob('*.py')
    dags = ['{p.parent.name}_{p.stem}'.format(p=p) for p in dag_files]
    dags = [d for d in dags if (d != DAG_ID)]
    execution_date = (datetime.date(datetime.now()) - timedelta(days=DAYS_TO_RETAIN))
    p = Path(LOG_DIR)
    for d in dags:
        logger.info('Cleaning up meta tables for {}'.format(d))
        for t in TABLES:
            sql = "delete from {} where dag_id='{}' and execution_date < '{}'".format(t, d, execution_date)
            hook.run(sql, True)
        logger.info('Cleaning up log folder for {}'.format(d))
        for path in list(p.glob('{}/*/*'.format(d))):
            log_date = str(path).split('/')[(- 1)]
            log_date = log_date.split('T')[0]
            log_date = datetime.date(datetime.strptime(log_date, '%Y-%m-%d'))
            if (log_date < execution_date):
                logger.info('Deleting dir {}'.format(str(path.absolute())))
                shutil.rmtree(str(path.absolute()))