A Django app for effortless download, restore, and loading of your server's database, media files, and state data as a zip file, preserving and restoring server states.
"""
ASGI config for DjangoStateKeeper project.

It exposes the ASGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/4.2/howto/deployment/asgi/
"""

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'DjangoStateKeeper.settings')

application = get_asgi_application()

DjangoStateKeeper/asgi.py
MMDBadCoder/Django-State-Keeper
"""
Django settings for DjangoStateKeeper project.

Generated by 'django-admin startproject' using Django 4.2.5.

For more information on this file, see
https://docs.djangoproject.com/en/4.2/topics/settings/

For the full list of settings and their values, see
https://docs.djangoproject.com/en/4.2/ref/settings/
"""

import os
from pathlib import Path

# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent

# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/4.2/howto/deployment/checklist/

# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = 'django-insecure-tne)$jcvln$q&&tv=w@f6n7&3&26md_&%$t86h$$#+i0=bnv=)'

# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True

ALLOWED_HOSTS = ['*']

# Application definition

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'django_state_keeper'
]

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

ROOT_URLCONF = 'DjangoStateKeeper.urls'

TEMPLATES = [
    {
        'BACKEND': 'django.template.backends.django.DjangoTemplates',
        'DIRS': [BASE_DIR / 'templates']
        ,
        'APP_DIRS': True,
        'OPTIONS': {
            'context_processors': [
                'django.template.context_processors.debug',
                'django.template.context_processors.request',
                'django.contrib.auth.context_processors.auth',
                'django.contrib.messages.context_processors.messages',
            ],
        },
    },
]

WSGI_APPLICATION = 'DjangoStateKeeper.wsgi.application'

# Database
# https://docs.djangoproject.com/en/4.2/ref/settings/#databases

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}

# Password validation
# https://docs.djangoproject.com/en/4.2/ref/settings/#auth-password-validators

AUTH_PASSWORD_VALIDATORS = [
    {
        'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
    },
    {
        'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
    },
    {
        'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
    },
    {
        'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
    },
]

# Internationalization
# https://docs.djangoproject.com/en/4.2/topics/i18n/

LANGUAGE_CODE = 'en-us'

TIME_ZONE = 'UTC'

USE_I18N = True

USE_TZ = True

# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/4.2/howto/static-files/

STATIC_URL = '/static/'
STATIC_PUBLIC_URL = STATIC_URL
STATIC_DIR = os.path.join(BASE_DIR, 'static')
STATIC_PUBLIC_DIR = STATIC_DIR
STATICFILES_DIRS = [STATIC_DIR, ]
STATIC_DIR = os.path.join(BASE_DIR, 'static')

MEDIA_DIR = os.path.join(BASE_DIR, 'media')
MEDIA_ROOT = MEDIA_DIR
MEDIA_URL = '/media/'

# Default primary key field type
# https://docs.djangoproject.com/en/4.2/ref/settings/#default-auto-field

DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'

DjangoStateKeeper/settings.py
MMDBadCoder/Django-State-Keeper
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('backup/', include('django_state_keeper.urls')),
    path('admin/', admin.site.urls),
]

DjangoStateKeeper/urls.py
MMDBadCoder/Django-State-Keeper
"""
WSGI config for DjangoStateKeeper project.

It exposes the WSGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/4.2/howto/deployment/wsgi/
"""

import os

from django.core.wsgi import get_wsgi_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'DjangoStateKeeper.settings')

application = get_wsgi_application()

DjangoStateKeeper/wsgi.py
MMDBadCoder/Django-State-Keeper
from django.contrib import admin
from django.urls import reverse
from django.utils.safestring import mark_safe
from django.urls.exceptions import NoReverseMatch
from django_state_keeper.models.auto_backup import AutoBackupService
from django_state_keeper.models.loading_backup import BackupUpload
from django_state_keeper.models.packaging_backup import BackupPackaging


class AutoBackupAdmin(admin.ModelAdmin):
    list_display = ('name', 'id')
    ordering = ['id']


admin.site.register(AutoBackupService, AutoBackupAdmin)


class BackupPackagingAdmin(admin.ModelAdmin):
    list_display = ('name', 'id', 'download_button')
    ordering = ['id']

    def download_button(self, movie):
        try:
            url = reverse('admin-download-backup', args=[movie.id])
        except NoReverseMatch:
            raise Exception("You should include django-state-keeper.urls to your project urls!"
                            "\nAdd something like this in your urls.py file:"
                            "\n\"urlpatterns += [path('your-customized-path/', include('django_state_keeper.urls'))]\"")

        button_html = f'<a class="button" href="{url}" target="_blank">Download</a>'
        return mark_safe(button_html)

    download_button.short_description = 'Download Backup'


admin.site.register(BackupPackaging, BackupPackagingAdmin)


class BackupUploadAdmin(admin.ModelAdmin):
    pass


admin.site.register(BackupUpload, BackupUploadAdmin)

django_state_keeper/admin.py
MMDBadCoder/Django-State-Keeper
from django.apps import AppConfig


class DjangoStateKeeperConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'django_state_keeper'

django_state_keeper/apps.py
MMDBadCoder/Django-State-Keeper
import logging

from django.conf import settings


class LoggerFactory:
    __instance = None

    @staticmethod
    def get_instance():
        if LoggerFactory.__instance is None:
            LoggerFactory.__instance = LoggerFactory.create_logger()
        return LoggerFactory.__instance

    @staticmethod
    def create_logger():
        log_level = logging.INFO
        if hasattr(settings, 'LOG_LEVEL'):
            log_level = settings.LOG_LEVEL
        # Create console handler with a higher log level
        console_handler = logging.StreamHandler()
        console_handler.setLevel(log_level)
        console_handler.setFormatter(CustomLogFormatter())

        logger = logging.getLogger()
        logger.setLevel(logging.DEBUG)
        logger.addHandler(console_handler)

        return logger


class CustomLogFormatter(logging.Formatter):
    white = '\x1b[38;5;255m'
    grey = "\x1b[38;20m"
    yellow = "\x1b[38;5;227m"
    red = "\x1b[38;5;9m"
    bold_red = "\x1b[31;1m"
    blue = '\x1b[38;5;45m'
    reset = "\x1b[0m"
    format = "%(asctime)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"

    FORMATS = {
        logging.DEBUG: white + format,
        logging.INFO: blue + format,
        logging.WARNING: yellow + format,
        logging.ERROR: red + format,
        logging.CRITICAL: bold_red + format
    }

    def format(self, record):
        log_fmt = self.FORMATS.get(record.levelno)
        formatter = logging.Formatter(log_fmt)
        return formatter.format(record)

django_state_keeper/logging_utils.py
MMDBadCoder/Django-State-Keeper
# Generated by Django 4.2.5 on 2023-09-29 19:08

from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    initial = True

    dependencies = [
    ]

    operations = [
        migrations.CreateModel(
            name='BackupPackaging',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('name', models.CharField(max_length=100)),
                ('paths_to_gather', models.TextField(default='db.sqlite', max_length=1000)),
            ],
        ),
        migrations.CreateModel(
            name='BackupUpload',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('file', models.FileField(blank=True, upload_to='loaded_backup/')),
            ],
        ),
        migrations.CreateModel(
            name='AutoBackupService',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('name', models.CharField(max_length=100)),
                ('period_minutes', models.IntegerField(default=86400)),
                ('bot_token', models.CharField(max_length=100)),
                ('chat_id', models.CharField(max_length=50)),
                ('packaging', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='django_state_keeper.backuppackaging')),
            ],
        ),
    ]

django_state_keeper/migrations/0001_initial.py
MMDBadCoder/Django-State-Keeper
# Generated by Django 4.2.5 on 2023-09-29 20:41

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('django_state_keeper', '0001_initial'),
    ]

    operations = [
        migrations.DeleteModel(
            name='BackupUpload',
        ),
        migrations.AlterField(
            model_name='backuppackaging',
            name='paths_to_gather',
            field=models.TextField(default='db.sqlite3', max_length=1000),
        ),
    ]

django_state_keeper/migrations/0002_delete_backupupload_and_more.py
MMDBadCoder/Django-State-Keeper
# Generated by Django 4.2.5 on 2023-09-29 20:49

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('django_state_keeper', '0002_delete_backupupload_and_more'),
    ]

    operations = [
        migrations.CreateModel(
            name='BackupUpload',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('zip_file', models.FileField(blank=True, upload_to='loaded_backup/')),
            ],
        ),
    ]

django_state_keeper/migrations/0003_backupupload.py
MMDBadCoder/Django-State-Keeper
import os
import random
import string
import sys
import threading
from datetime import datetime
from threading import Thread
from time import sleep

import telegram
from asgiref.sync import async_to_sync
from django.db import models
from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver

from django_state_keeper.logging_utils import LoggerFactory
from django_state_keeper.models.packaging_backup import BackupPackaging


class AutoBackupService(models.Model):
    THREADS_BY_ID = {}
    SHOULD_CONTINUE_BY_THREAD_NAME = {}
    name = models.CharField(max_length=100, blank=False, null=False)
    packaging = models.ForeignKey(BackupPackaging, blank=False, null=False, on_delete=models.CASCADE)
    period_minutes = models.IntegerField(default=24 * 60 * 60, blank=False, null=False)
    bot_token = models.CharField(max_length=100, blank=False, null=False)
    chat_id = models.CharField(max_length=50, blank=False, null=False)

    class Meta:
        verbose_name = 'Auto Backup Sender Service'
        verbose_name_plural = 'Auto Backup Sender Services'


@receiver(post_save, sender=AutoBackupService)
def create_backup_thread(sender, instance: AutoBackupService, **kwargs):
    stop_last_thread_of_backup_service(instance)
    create_new_thread_of_backup_service(instance)


@receiver(pre_delete, sender=AutoBackupService)
def delete_backup_thread(sender, instance, **kwargs):
    stop_last_thread_of_backup_service(instance)


def create_new_thread_of_backup_service(instance: AutoBackupService):
    new_thread_name = f'backup-sender-{instance.id}' + generate_random_string(10)
    AutoBackupService.SHOULD_CONTINUE_BY_THREAD_NAME[new_thread_name] = True
    new_thread = Thread(name=new_thread_name, target=backup_sender_thread, args=(instance,))
    AutoBackupService.THREADS_BY_ID[instance.id] = new_thread
    new_thread.start()
    LoggerFactory.get_instance().info(f'Thread of {instance.name} backup service was created.')


def stop_last_thread_of_backup_service(instance: AutoBackupService):
    if AutoBackupService.THREADS_BY_ID.__contains__(instance.id):
        stopping_thread: Thread = AutoBackupService.THREADS_BY_ID[instance.id]
        AutoBackupService.SHOULD_CONTINUE_BY_THREAD_NAME[stopping_thread.name] = False
        LoggerFactory.get_instance().info(f'Thread of {instance.name} backup service will no longer continue.')


def backup_sender_thread(instance: AutoBackupService):
    while AutoBackupService.SHOULD_CONTINUE_BY_THREAD_NAME[threading.current_thread().name]:
        try:
            send_backup_in_telegram(instance)
        except Exception as e:
            LoggerFactory.get_instance().error(e)
        sleep(60 * instance.period_minutes)
    LoggerFactory.get_instance().info(f'Thread of {instance.name} backup service ended.')


def send_backup_in_telegram(instance: AutoBackupService):
    bot = telegram.Bot(token=instance.bot_token)
    package_path = instance.packaging.create_a_backup_package()
    caption = f'{instance.name}\n{datetime.now().strftime("%Y/%m/%d %H:%M:%S")}'
    with open(package_path, 'rb') as file:
        async def async_def():
            await bot.send_document(chat_id=instance.chat_id, document=file, caption=caption)

        async_to_sync(async_def)()
    os.remove(path=package_path)


def generate_random_string(length):
    characters = string.ascii_letters + string.digits + string.punctuation
    return ''.join(random.choice(characters) for _ in range(length))


if len(sys.argv) > 1 and sys.argv[1] == 'runserver':
    for auto_backup in AutoBackupService.objects.all():
        create_new_thread_of_backup_service(auto_backup)

django_state_keeper/models/auto_backup.py
MMDBadCoder/Django-State-Keeper
import os
import zipfile

from django.conf import settings
from django.db import models
from django.db.models.signals import post_save
from django.dispatch import receiver


class BackupUpload(models.Model):
    zip_file = models.FileField(upload_to='loaded_backup/', blank=True)

    class Meta:
        verbose_name = 'Upload Backup'
        verbose_name_plural = 'Upload Backup'

@receiver(post_save, sender=BackupUpload)
def create_backup_thread(sender, instance: BackupUpload, **kwargs):
    with zipfile.ZipFile(instance.zip_file.path, 'r') as zip_ref:
        zip_ref.extractall(settings.BASE_DIR)
    instance.delete()
    os.remove(instance.zip_file.path)

django_state_keeper/models/loading_backup.py
MMDBadCoder/Django-State-Keeper
import os
import shutil
import sys
import tempfile
import threading
from datetime import datetime, timedelta
from time import sleep

from django.conf import settings
from django.core.exceptions import BadRequest
from django.db import models

from django_state_keeper.logging_utils import LoggerFactory


class BackupPackaging(models.Model):
    TIME_OF_ZIP_CREATION = {}
    name = models.CharField(max_length=100, blank=False, null=False)
    paths_to_gather = models.TextField(max_length=1000, blank=False, null=False, default='db.sqlite3')

    def __str__(self):
        return self.name

    def create_a_backup_package(self):
        temp_dir = tempfile.mkdtemp()
        try:
            # Copy pase mentioned files and directories to a temporary directory
            for relative_source_path in self.paths_to_gather.split('\n'):
                relative_source_path = relative_source_path.strip()
                if not relative_source_path:
                    continue
                absolute_source_path = os.path.join(settings.BASE_DIR, relative_source_path)
                if os.path.isfile(absolute_source_path):
                    absolute_target_path = os.path.join(temp_dir, relative_source_path)
                    os.makedirs(os.path.dirname(absolute_target_path), exist_ok=True)
                    shutil.copy(absolute_source_path, absolute_target_path)
                elif os.path.isdir(absolute_source_path):
                    absolute_target_directory = os.path.join(temp_dir, relative_source_path)
                    shutil.copytree(absolute_source_path, absolute_target_directory)
                else:
                    LoggerFactory.get_instance().error("Invalid source path:", absolute_source_path)

            # Create a zip file containing all the files in the temporary directory
            zip_file_name = f'{self.name.replace(" ", "_")}-{datetime.now().strftime("%Y-%m-%d_%H-%M-%S")}'
            creating_zip_path = os.path.join(tempfile.tempdir, zip_file_name)
            creating_zip_path = shutil.make_archive(creating_zip_path, 'zip', temp_dir)
            BackupPackaging.TIME_OF_ZIP_CREATION[creating_zip_path] = datetime.now()
            LoggerFactory.get_instance().info(f'{creating_zip_path} was created.')
            return creating_zip_path
        except Exception as e:
            raise BadRequest(e)


def created_zips_cleaner():
    hour_difference = timedelta(hours=1)
    while True:
        for created_zip_path, creation_time in BackupPackaging.TIME_OF_ZIP_CREATION.items():
            if datetime.now() - creation_time > hour_difference:
                try:
                    os.remove(created_zip_path)
                    del BackupPackaging.TIME_OF_ZIP_CREATION[created_zip_path]
                    LoggerFactory.get_instance().info(f'{created_zip_path} was deleted.')
                except Exception as e:
                    LoggerFactory.get_instance().error(e)
        sleep(hour_difference.seconds)


if len(sys.argv) > 1 and sys.argv[1] == 'runserver':
    threading.Thread(target=created_zips_cleaner).start()

django_state_keeper/models/packaging_backup.py
MMDBadCoder/Django-State-Keeper
from django.urls import path

from . import views

urlpatterns = [
    path('download/<int:packaging_id>/', views.download_file_view, name='admin-download-backup'),
]
django_state_keeper/urls.py
MMDBadCoder/Django-State-Keeper
from django.contrib.auth.decorators import login_required, user_passes_test
from django.http import FileResponse

from django_state_keeper.models.packaging_backup import BackupPackaging


@login_required
@user_passes_test(lambda u: u.is_superuser)
def download_file_view(request, packaging_id):
    packaging_instance = BackupPackaging.objects.get(id=packaging_id)
    created_zip_path = packaging_instance.create_a_backup_package()
    response = FileResponse(open(created_zip_path, 'rb'))
    response['Content-Disposition'] = 'attachment; filename="{}"'.format(created_zip_path)
    return response

django_state_keeper/views.py
MMDBadCoder/Django-State-Keeper
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys


def main():
    """Run administrative tasks."""
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'DjangoStateKeeper.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)


if __name__ == '__main__':
    main()

manage.py
MMDBadCoder/Django-State-Keeper
from setuptools import setup, find_packages

setup(
    name='django-state-keeper',
    version='1.1',
    packages=find_packages(),
    include_package_data=True,
    license='MIT',
    description="A Django app that allows you to create backup of project and restore it later.",
    long_description_content_type='text/markdown',
    long_description=open('README.md').read(),
    url='https://github.com/MMDBadCoder/Django-State-Keeper',
    author='Mohammad Heydari',
    author_email='heidary13794@gmail.com',
    classifiers=[
        'Development Status :: 5 - Production/Stable',
        'Environment :: Web Environment',
        'Framework :: Django',
        'Framework :: Django :: 3.2',
        'Intended Audience :: Developers',
        'License :: OSI Approved :: MIT License',
        'Operating System :: OS Independent',
        'Programming Language :: Python',
        'Programming Language :: Python :: 3',
        'Programming Language :: Python :: 3.9',
        'Topic :: Internet :: WWW/HTTP',
        'Topic :: Internet :: WWW/HTTP :: Dynamic Content',
        'Topic :: Software Development :: Libraries',
        'Topic :: Software Development :: Libraries :: Application Frameworks',
        'Topic :: Software Development :: Libraries :: Python Modules',
    ],
    install_requires=[
        'Django>=3.2',
        'python-telegram-bot>=20.5'
    ],
)

setup.py
MMDBadCoder/Django-State-Keeper