<filename>dags/full_backup_example.py
from airflow import DAG
from airflow.models import Variable
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta

# Add these variables in admin panel, 
# Menu: admin/Variables
# 
### Where to backup, E.g:
# key:full_backup_example_backup_path 
# value: "/backup/path_to_full_backup/"
backup_path =  Variable.get('full_backup_example_backup_path')
# If disk usage is morethan this variable value (percentage), this dag will be fail
# key: full_backup_example_max_allow_disk_usage
# value: 90
max_allow_disk_usage =  Variable.get("full_backup_example_max_allow_disk_usage")
# To delete backup files older than this variable value (days)
# key: full_backup_example_old_backup_age
# value: 7
old_backup_age = Variable.get("full_backup_example_old_backup_age")
# Run this dag with this variable value period
# E.g Every three days at 2:00 am
# key:full_backup_example_schedule_interval
# value: "0 2 */3 * *"
schedule_interval =  Variable.get("full_backup_example_schedule_interval")

### Database credentials
# database username
backup_name =  Variable.get("full_backup_example_backup_name")
db_user =  Variable.get("full_backup_example_db_user")
db_pass =  Variable.get("full_backup_example_db_pass")
db_name =  Variable.get("full_backup_example_db_name")
db_addr =  Variable.get("full_backup_example_db_addr")




with DAG(
    dag_id='full_backup_example', 
    schedule_interval=schedule_interval, 
    start_date=datetime(2021, 1, 1), 
    catchup=False
    ) as dag:

# Delete old backups
disk_cleanup_script="""/opt/airflow/scripts/delete_old_backups.sh {} 7""".format(backup_path)
disk_cleanup = BashOperator(
    task_id='disk_cleanup',
    bash_command=disk_cleanup_script
    )

# Check disk usage before start
check_disk_usage_script="""/opt/airflow/scripts/check_disk_usage.sh {} {}""".format(backup_path, max_allow_disk_usage)
check_disk_usage = BashOperator(
    task_id='disk_usage',
    bash_command=check_disk_usage_script
    )

# Run full backup
run_backup_script="""/opt/airflow/scripts/mysql/full_backup.sh {} {} {} {} {} {}""".format(
                    backup_path,
                    db_user,
                    db_pass,
                    db_name,
                    db_addr,
                    backup_name
                    )
run_backup = BashOperator(
    task_id='bash_task', 
    bash_command="")


disk_cleanup >> check_disk_usage >> run_backup

<filename>dags/full_backup_example.py
from airflow import DAG
from airflow.models import Variable
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta

# Add these variables in admin panel, 
# Menu: admin/Variables
# 
### Where to backup, E.g:
# key:full_backup_example_backup_path 
# value: "/backup/path_to_full_backup/"
backup_path =  Variable.get('full_backup_example_backup_path')
# If disk usage is morethan this variable value (percentage), this dag will be fail
# key: full_backup_example_max_allow_disk_usage
# value: 90
max_allow_disk_usage =  Variable.get("full_backup_example_max_allow_disk_usage")
# To delete backup files older than this variable value (days)
# key: full_backup_example_old_backup_age
# value: 7
old_backup_age = Variable.get("full_backup_example_old_backup_age")
# Run this dag with this variable value period
# E.g Every three days at 2:00 am
# key:full_backup_example_schedule_interval
# value: "0 2 */3 * *"
schedule_interval =  Variable.get("full_backup_example_schedule_interval")

### Database credentials
# database username
backup_name =  Variable.get("full_backup_example_backup_name")
db_user =  Variable.get("full_backup_example_db_user")
db_pass =  Variable.get("full_backup_example_db_pass")
db_name =  Variable.get("full_backup_example_db_name")
db_addr =  Variable.get("full_backup_example_db_addr")




with DAG(
    dag_id='full_backup_example', 
    schedule_interval=schedule_interval, 
    start_date=datetime(2021, 1, 1), 
    catchup=False
    ) as dag:

# Delete old backups
disk_cleanup_script="""/opt/airflow/scripts/delete_old_backups.sh {} 7""".format(backup_path)
disk_cleanup = BashOperator(
    task_id='disk_cleanup',
    bash_command=disk_cleanup_script
    )

# Check disk usage before start
check_disk_usage_script="""/opt/airflow/scripts/check_disk_usage.sh {} {}""".format(backup_path, max_allow_disk_usage)
check_disk_usage = BashOperator(
    task_id='disk_usage',
    bash_command=check_disk_usage_script
    )

# Run full backup
run_backup_script="""/opt/airflow/scripts/mysql/full_backup.sh {} {} {} {} {} {}""".format(
                    backup_path,
                    db_user,
                    db_pass,
                    db_name,
                    db_addr,
                    backup_name
                    )
run_backup = BashOperator(
    task_id='bash_task', 
    bash_command="")


disk_cleanup >> check_disk_usage >> run_backup

# Add these variables in admin panel, # Menu: admin/Variables # ### Where to backup, E.g: # key:full_backup_example_backup_path # value: "/backup/path_to_full_backup/" # If disk usage is morethan this variable value (percentage), this dag will be fail # key: full_backup_example_max_allow_disk_usage # value: 90 # To delete backup files older than this variable value (days) # key: full_backup_example_old_backup_age # value: 7 # Run this dag with this variable value period # E.g Every three days at 2:00 am # key:full_backup_example_schedule_interval # value: "0 2 */3 * *" ### Database credentials # database username # Delete old backups /opt/airflow/scripts/delete_old_backups.sh {} 7 # Check disk usage before start /opt/airflow/scripts/check_disk_usage.sh {} {} # Run full backup /opt/airflow/scripts/mysql/full_backup.sh {} {} {} {} {} {}