nestict/ihrm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Pylint updates

Browse Source
This commit is contained in:
Horilla
2025-02-14 10:01:07 +05:30
parent 855258838d
commit 07e9c2331e
16 changed files with 273 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
horilla/locale/pt-br/LC_MESSAGES/django.po
View File

@@ -25608,4 +25608,3 @@ msgstr "Progresso"
#~ msgid "individual"
#~ msgstr "Creation"

2
horilla_backup/__init__.py
View File

@@ -1 +1 @@
default_app_config = 'horilla_backup.apps.backupConfig'
default_app_config = "horilla_backup.apps.backupConfig"

3
horilla_backup/admin.py
View File

@@ -1,9 +1,8 @@
from django.contrib import admin
from .models import *
# Register your models here.
admin.site.register(LocalBackup)
admin.site.register(GoogleDriveBackup)

7
horilla_backup/apps.py
View File

@@ -1,15 +1,16 @@
from django.apps import AppConfig
class BackupConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'horilla_backup'
default_auto_field = "django.db.models.BigAutoField"
name = "horilla_backup"
def ready(self):
from django.urls import include, path
from horilla.urls import urlpatterns
urlpatterns.append(
path("backup/", include("horilla_backup.urls")),
)
super().ready()

152
horilla_backup/forms.py
View File

@@ -1,27 +1,38 @@
from django import forms
from .models import *
from base.forms import ModelForm
from django.template.loader import render_to_string
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
from pathlib import Path
from .gdrive import authenticate
from django.core.files.storage import default_storage
from django.core.files.base import ContentFile
import os
from pathlib import Path
from django import forms
from django.core.exceptions import ValidationError
from django.core.files.base import ContentFile
from django.core.files.storage import default_storage
from django.template.loader import render_to_string
from django.utils.translation import gettext_lazy as _
from base.forms import ModelForm
from .gdrive import authenticate
from .models import *
class LocalBackupSetupForm(ModelForm):
verbose_name = "Server Backup"
backup_db = forms.BooleanField(required=False, help_text="Enable to backup database to server.")
backup_media = forms.BooleanField(required=False, help_text="Enable to backup all media files to server.")
interval = forms.BooleanField(required=False, help_text="Enable to automate the backup in a period of seconds.")
fixed = forms.BooleanField(required=False, help_text="Enable to automate the backup in a fixed time.")
backup_db = forms.BooleanField(
required=False, help_text="Enable to backup database to server."
)
backup_media = forms.BooleanField(
required=False, help_text="Enable to backup all media files to server."
)
interval = forms.BooleanField(
required=False,
help_text="Enable to automate the backup in a period of seconds.",
)
fixed = forms.BooleanField(
required=False, help_text="Enable to automate the backup in a fixed time."
)
class Meta:
model = LocalBackup
exclude = ['active']
exclude = ["active"]
def as_p(self):
"""
@@ -31,63 +42,64 @@ class LocalBackupSetupForm(ModelForm):
table_html = render_to_string("common_form.html", context)
return table_html
def clean(self):
cleaned_data = super().clean()
backup_db = cleaned_data.get('backup_db')
backup_media = cleaned_data.get('backup_media')
interval = cleaned_data.get('interval')
fixed = cleaned_data.get('fixed')
seconds = cleaned_data.get('seconds')
hour = cleaned_data.get('hour')
minute = cleaned_data.get('minute')
backup_path = cleaned_data.get('backup_path')
backup_db = cleaned_data.get("backup_db")
backup_media = cleaned_data.get("backup_media")
interval = cleaned_data.get("interval")
fixed = cleaned_data.get("fixed")
seconds = cleaned_data.get("seconds")
hour = cleaned_data.get("hour")
minute = cleaned_data.get("minute")
backup_path = cleaned_data.get("backup_path")
path = Path(backup_path)
if not path.exists():
raise ValidationError({
'backup_path': _('The directory does not exist.')
})
raise ValidationError({"backup_path": _("The directory does not exist.")})
if backup_db == False and backup_media == False:
raise forms.ValidationError("Please select any backup option.")
if interval == False and fixed == False:
raise forms.ValidationError("Please select any backup automate option.")
if interval == True and seconds == None:
raise ValidationError({
'seconds': _('This field is required.')
})
raise ValidationError({"seconds": _("This field is required.")})
if fixed == True and hour == None:
raise ValidationError({
'hour': _('This field is required.')
})
raise ValidationError({"hour": _("This field is required.")})
if seconds:
if seconds < 0:
raise ValidationError({
'seconds': _('Negative value is not accepatable.')
})
raise ValidationError(
{"seconds": _("Negative value is not accepatable.")}
)
if hour:
if hour < 0 or hour > 24:
raise ValidationError({
'hour': _('Enter a hour between 0 to 24.')
})
raise ValidationError({"hour": _("Enter a hour between 0 to 24.")})
if minute:
if minute < 0 or minute > 60:
raise ValidationError({
'minute': _('Enter a minute between 0 to 60.')
})
raise ValidationError({"minute": _("Enter a minute between 0 to 60.")})
return cleaned_data
class GdriveBackupSetupForm(ModelForm):
verbose_name = "Gdrive Backup"
backup_db = forms.BooleanField(required=False, label="Backup DB", help_text="Enable to backup database to Gdrive")
backup_media = forms.BooleanField(required=False, label="Backup Media", help_text="Enable to backup all media files to Gdrive")
interval = forms.BooleanField(required=False, help_text="Enable to automate the backup in a period of seconds.")
fixed = forms.BooleanField(required=False, help_text="Enable to automate the backup in a fixed time.")
backup_db = forms.BooleanField(
required=False,
label="Backup DB",
help_text="Enable to backup database to Gdrive",
)
backup_media = forms.BooleanField(
required=False,
label="Backup Media",
help_text="Enable to backup all media files to Gdrive",
)
interval = forms.BooleanField(
required=False,
help_text="Enable to automate the backup in a period of seconds.",
)
fixed = forms.BooleanField(
required=False, help_text="Enable to automate the backup in a fixed time."
)
class Meta:
model = GoogleDriveBackup
exclude = ['active']
exclude = ["active"]
def as_p(self):
"""
@@ -99,14 +111,14 @@ class GdriveBackupSetupForm(ModelForm):
def clean(self):
cleaned_data = super().clean()
backup_db = cleaned_data.get('backup_db')
backup_media = cleaned_data.get('backup_media')
interval = cleaned_data.get('interval')
fixed = cleaned_data.get('fixed')
seconds = cleaned_data.get('seconds')
hour = cleaned_data.get('hour')
minute = cleaned_data.get('minute')
service_account_file = cleaned_data.get('service_account_file')
backup_db = cleaned_data.get("backup_db")
backup_media = cleaned_data.get("backup_media")
interval = cleaned_data.get("interval")
fixed = cleaned_data.get("fixed")
seconds = cleaned_data.get("seconds")
hour = cleaned_data.get("hour")
minute = cleaned_data.get("minute")
service_account_file = cleaned_data.get("service_account_file")
try:
if GoogleDriveBackup.objects.exists():
@@ -117,7 +129,9 @@ class GdriveBackupSetupForm(ModelForm):
file_name = service_account_file.name
new_file_name = file_name
# Save using Django's default storage system
relative_path = default_storage.save(new_file_name, ContentFile(file_data))
relative_path = default_storage.save(
new_file_name, ContentFile(file_data)
)
# Get the full absolute path
full_path = default_storage.path(relative_path)
authenticate(full_path)
@@ -130,26 +144,18 @@ class GdriveBackupSetupForm(ModelForm):
if interval == False and fixed == False:
raise forms.ValidationError("Please select any backup automate option.")
if interval == True and seconds == None:
raise ValidationError({
'seconds': _('This field is required.')
})
raise ValidationError({"seconds": _("This field is required.")})
if fixed == True and hour == None:
raise ValidationError({
'hour': _('This field is required.')
})
raise ValidationError({"hour": _("This field is required.")})
if seconds:
if seconds < 0:
raise ValidationError({
'seconds': _('Negative value is not accepatable.')
})
raise ValidationError(
{"seconds": _("Negative value is not accepatable.")}
)
if hour:
if hour < 0 or hour > 24:
raise ValidationError({
'hour': _('Enter a hour between 0 to 24.')
})
raise ValidationError({"hour": _("Enter a hour between 0 to 24.")})
if minute:
if minute < 0 or minute > 60:
raise ValidationError({
'minute': _('Enter a minute between 0 to 60.')
})
raise ValidationError({"minute": _("Enter a minute between 0 to 60.")})
return cleaned_data

29
horilla_backup/gdrive.py
View File

@@ -1,29 +1,28 @@
from googleapiclient.discovery import build
from google.oauth2 import service_account
from googleapiclient.http import MediaFileUpload
import os
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
SCOPES = ['https://www.googleapis.com/auth/drive']
SCOPES = ["https://www.googleapis.com/auth/drive"]
def authenticate(service_account_file):
creds = service_account.Credentials.from_service_account_file(service_account_file, scopes=SCOPES)
creds = service_account.Credentials.from_service_account_file(
service_account_file, scopes=SCOPES
)
return creds
def upload_file(file_path, service_account_file, parent_folder_id):
creds = authenticate(service_account_file)
service = build('drive', 'v3', credentials=creds)
service = build("drive", "v3", credentials=creds)
parent_folder_id = parent_folder_id
file_metadata = {
'name' : os.path.basename(file_path),
'parents' : [parent_folder_id]
}
file_metadata = {"name": os.path.basename(file_path), "parents": [parent_folder_id]}
media = MediaFileUpload(file_path, resumable=True)
file = service.files().create(
body=file_metadata,
media_body=media,
fields='id'
).execute()
file = (
service.files()
.create(body=file_metadata, media_body=media, fields="id")
.execute()
)

2
horilla_backup/migrations/__init__.py
View File

@@ -3,6 +3,7 @@ import atexit
def shutdown_function():
from horilla_backup.models import GoogleDriveBackup, LocalBackup
if GoogleDriveBackup.objects.exists():
google_drive_backup = GoogleDriveBackup.objects.first()
google_drive_backup.active = False
@@ -12,6 +13,7 @@ def shutdown_function():
local_backup.active = False
local_backup.save()
try:
atexit.register(shutdown_function)
except:

26
horilla_backup/models.py
View File

@@ -1,9 +1,13 @@
from django.db import models
#Create your models here.
# Create your models here.
class LocalBackup(models.Model):
backup_path = models.CharField(max_length=255, help_text="Specify the path in the server were the backup files should keep")
backup_path = models.CharField(
max_length=255,
help_text="Specify the path in the server were the backup files should keep",
)
backup_media = models.BooleanField(blank=True, null=True)
backup_db = models.BooleanField(blank=True, null=True)
interval = models.BooleanField(blank=True, null=True)
@@ -13,7 +17,6 @@ class LocalBackup(models.Model):
minute = models.IntegerField(blank=True, null=True)
active = models.BooleanField(default=False)
def save(self, *args, **kwargs):
# Check if there's an existing instance
if self.interval == False:
@@ -26,7 +29,7 @@ class LocalBackup(models.Model):
existing_instance = LocalBackup.objects.first()
# Update the fields of the existing instance with the new data
for field in self._meta.fields:
if field.name != 'id': # Avoid changing the primary key
if field.name != "id": # Avoid changing the primary key
setattr(existing_instance, field.name, getattr(self, field.name))
# Save the updated instance
super(LocalBackup, existing_instance).save(*args, **kwargs)
@@ -38,12 +41,16 @@ class LocalBackup(models.Model):
class GoogleDriveBackup(models.Model):
service_account_file = models.FileField(upload_to="gdrive_service_account_file",
service_account_file = models.FileField(
upload_to="gdrive_service_account_file",
verbose_name="Service Account File",
help_text="Make sure your file is in JSON format and contains your Google Service Account credentials")
gdrive_folder_id = models.CharField(max_length=255,
help_text="Make sure your file is in JSON format and contains your Google Service Account credentials",
)
gdrive_folder_id = models.CharField(
max_length=255,
verbose_name="Gdrive Folder ID",
help_text="Shared Gdrive folder Id with access granted to Gmail service account. Enable full permissions for seamless connection.")
help_text="Shared Gdrive folder Id with access granted to Gmail service account. Enable full permissions for seamless connection.",
)
backup_media = models.BooleanField(blank=True, null=True)
backup_db = models.BooleanField(blank=True, null=True)
interval = models.BooleanField(blank=True, null=True)
@@ -53,7 +60,6 @@ class GoogleDriveBackup(models.Model):
minute = models.IntegerField(blank=True, null=True)
active = models.BooleanField(default=False)
def save(self, *args, **kwargs):
# Check if there's an existing instance
if self.interval == False:
@@ -66,7 +72,7 @@ class GoogleDriveBackup(models.Model):
existing_instance = GoogleDriveBackup.objects.first()
# Update the fields of the existing instance with the new data
for field in self._meta.fields:
if field.name != 'id': # Avoid changing the primary key
if field.name != "id": # Avoid changing the primary key
setattr(existing_instance, field.name, getattr(self, field.name))
# Save the updated instance
super(GoogleDriveBackup, existing_instance).save(*args, **kwargs)

36
horilla_backup/pgdump.py
View File

@@ -1,30 +1,38 @@
import subprocess
import os
import subprocess
def dump_postgres_db(db_name, username, output_file, password=None, host='localhost', port=5432):
def dump_postgres_db(
db_name, username, output_file, password=None, host="localhost", port=5432
):
# Set environment variable for the password if provided
if password:
os.environ['PGPASSWORD'] = password
os.environ["PGPASSWORD"] = password
# Construct the pg_dump command
dump_command = [
'pg_dump',
'-h', host,
'-p', str(port),
'-U', username,
'-F', 'c', # Custom format
'-f', output_file,
db_name
"pg_dump",
"-h",
host,
"-p",
str(port),
"-U",
username,
"-F",
"c", # Custom format
"-f",
output_file,
db_name,
]
try:
# Execute the pg_dump command
result = subprocess.run(dump_command, check=True, text=True, capture_output=True)
result = subprocess.run(
dump_command, check=True, text=True, capture_output=True
)
except subprocess.CalledProcessError as e:
pass
finally:
# Clean up the environment variable
if password:
del os.environ['PGPASSWORD']
del os.environ["PGPASSWORD"]

44
horilla_backup/scheduler.py
View File

@@ -1,11 +1,15 @@
import os
from apscheduler.schedulers.background import BackgroundScheduler
from django.core.management import call_command
import os
from horilla import settings
from .gdrive import *
# from horilla.settings import DBBACKUP_STORAGE_OPTIONS
from .models import *
from .gdrive import *
from .pgdump import *
from horilla import settings
from .zip import *
scheduler = BackgroundScheduler()
@@ -83,15 +87,15 @@ def google_drive_backup():
service_account_file = google_drive.service_account_file.path
gdrive_folder_id = google_drive.gdrive_folder_id
if google_drive.backup_db:
db = settings.DATABASES['default']
db = settings.DATABASES["default"]
dump_postgres_db(
db_name=db['NAME'],
username=db['USER'],
output_file='backupdb.dump',
password=db['PASSWORD']
db_name=db["NAME"],
username=db["USER"],
output_file="backupdb.dump",
password=db["PASSWORD"],
)
upload_file('backupdb.dump', service_account_file, gdrive_folder_id)
os.remove('backupdb.dump')
upload_file("backupdb.dump", service_account_file, gdrive_folder_id)
os.remove("backupdb.dump")
if google_drive.backup_media:
folder_to_zip = settings.MEDIA_ROOT
output_zip_file = "media.zip"
@@ -110,14 +114,25 @@ def start_gdrive_backup_job():
# Remove existing job if it exists
try:
scheduler.remove_job('backup_job')
scheduler.remove_job("backup_job")
except:
pass
# Add new job based on Gdrive Backup configuration
if gdrive_backup.interval:
scheduler.add_job(google_drive_backup, 'interval', seconds=gdrive_backup.seconds, id='gdrive_backup_job')
scheduler.add_job(
google_drive_backup,
"interval",
seconds=gdrive_backup.seconds,
id="gdrive_backup_job",
)
else:
scheduler.add_job(google_drive_backup, trigger='cron', hour=gdrive_backup.hour, minute=gdrive_backup.minute, id='gdrive_backup_job')
scheduler.add_job(
google_drive_backup,
trigger="cron",
hour=gdrive_backup.hour,
minute=gdrive_backup.minute,
id="gdrive_backup_job",
)
# Start the scheduler if it's not already running
if not scheduler.running:
@@ -132,7 +147,7 @@ def stop_gdrive_backup_job():
Stop the backup job if it exists.
"""
try:
scheduler.remove_job('gdrive_backup_job')
scheduler.remove_job("gdrive_backup_job")
except:
pass
@@ -143,4 +158,3 @@ def stop_gdrive_backup_job():
# """
# stop_gdrive_backup_job()
# start_gdrive_backup_job()

3
horilla_backup/urls.py
View File

@@ -1,4 +1,5 @@
from django.urls import path
from .views import *
urlpatterns = [
@@ -7,5 +8,5 @@ urlpatterns = [
# path("delete/", local_Backup_delete, name="backup_delete"),
path("gdrive/", gdrive_setup, name="gdrive"),
path("gdrive-start-stop/", gdrive_Backup_stop_or_start, name="gdrive_start_stop"),
path("gdrive-delete/", gdrive_Backup_delete, name="gdrive_delete")
path("gdrive-delete/", gdrive_Backup_delete, name="gdrive_delete"),
]

26
horilla_backup/views.py
View File

@@ -1,4 +1,8 @@
from django.shortcuts import render, redirect
from django.contrib import messages
from django.db import connection
from django.shortcuts import redirect, render
from django.utils.translation import gettext_lazy as _
from horilla.decorators import (
hx_request_required,
login_required,
@@ -6,15 +10,12 @@ from horilla.decorators import (
owner_can_enter,
permission_required,
)
from .forms import *
from django.contrib import messages
from django.utils.translation import gettext_lazy as _
from .scheduler import *
from .gdrive import *
from .pgdump import *
from .scheduler import *
from .zip import *
from django.db import connection
# @login_required
# @permission_required("backup.add_localbackup")
@@ -127,8 +128,11 @@ def gdrive_setup(request):
stop_gdrive_backup_job()
messages.success(request, _("gdrive backup automation setup updated."))
return redirect("gdrive")
return render(request, "backup/gdrive_setup_form.html", {"form": form, "show":show, "active":active})
return render(
request,
"backup/gdrive_setup_form.html",
{"form": form, "show": show, "active": active},
)
if request.method == "POST":
form = GdriveBackupSetupForm(request.POST, request.FILES)
@@ -136,7 +140,11 @@ def gdrive_setup(request):
form.save()
messages.success(request, _("gdrive backup automation setup Created."))
return redirect("gdrive")
return render(request, "backup/gdrive_setup_form.html", {"form": form, "show":show, "active":active})
return render(
request,
"backup/gdrive_setup_form.html",
{"form": form, "show": show, "active": active},
)
@login_required

6
horilla_backup/zip.py
View File

@@ -1,8 +1,9 @@
import os
import zipfile
def zip_folder(folder_path, output_zip_path):
with zipfile.ZipFile(output_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
with zipfile.ZipFile(output_zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
# Walk the directory
for root, dirs, files in os.walk(folder_path):
for file in files:
@@ -10,6 +11,3 @@ def zip_folder(folder_path, output_zip_path):
file_path = os.path.join(root, file)
# Add file to zip, preserving the folder structure
zipf.write(file_path, os.path.relpath(file_path, folder_path))