[UPDT] BASE: Generic pylint fixes and removal of unwanted statements

This commit is contained in:
Horilla
2025-02-03 14:47:32 +05:30
parent fbbdba4663
commit c8898e93c0
6 changed files with 214 additions and 200 deletions

View File

@@ -7,7 +7,6 @@ from datetime import datetime, timedelta
from django.contrib import messages from django.contrib import messages
from django.contrib.auth.models import User from django.contrib.auth.models import User
from django.core.paginator import EmptyPage, PageNotAnInteger, Paginator
from django.db.models import Q from django.db.models import Q
from django.http import HttpResponse from django.http import HttpResponse
from django.shortcuts import redirect, render from django.shortcuts import redirect, render
@@ -30,6 +29,13 @@ from notifications.signals import notify
@login_required @login_required
@hx_request_required @hx_request_required
def announcement_list(request): def announcement_list(request):
"""
Renders a list of announcements for the authenticated user.
This view fetches all announcements and updates their expiration dates if not already set.
It filters announcements based on the user's permissions and whether the announcements
are still valid (not expired). Additionally, it checks if the user has viewed each announcement.
"""
general_expire_date = ( general_expire_date = (
AnnouncementExpire.objects.values_list("days", flat=True).first() or 30 AnnouncementExpire.objects.values_list("days", flat=True).first() or 30
) )
@@ -45,25 +51,26 @@ def announcement_list(request):
if announcements_to_update: if announcements_to_update:
Announcement.objects.bulk_update(announcements_to_update, ["expire_date"]) Announcement.objects.bulk_update(announcements_to_update, ["expire_date"])
has_view_permission = request.user.has_perm("base.view_announcement")
announcements = announcements.filter(expire_date__gte=datetime.today().date()) announcements = announcements.filter(expire_date__gte=datetime.today().date())
announcement_items = (
if request.user.has_perm("base.view_announcement"): announcements
announcement_list = announcements if has_view_permission
else: else announcements.filter(
announcement_list = announcements.filter(
Q(employees=request.user.employee_get) | Q(employees__isnull=True) Q(employees=request.user.employee_get) | Q(employees__isnull=True)
) )
)
announcement_list = announcement_list.prefetch_related( filtered_announcements = announcement_items.prefetch_related(
"announcementview_set" "announcementview_set"
).order_by("-created_at") ).order_by("-created_at")
for announcement in announcement_list: for announcement in filtered_announcements:
announcement.has_viewed = announcement.announcementview_set.filter( announcement.has_viewed = announcement.announcementview_set.filter(
user=request.user, viewed=True user=request.user, viewed=True
).exists() ).exists()
instance_ids = json.dumps([instance.id for instance in announcement_list]) instance_ids = json.dumps([instance.id for instance in filtered_announcements])
context = { context = {
"announcements": announcement_list, "announcements": filtered_announcements,
"general_expire_date": general_expire_date, "general_expire_date": general_expire_date,
"instance_ids": instance_ids, "instance_ids": instance_ids,
} }

View File

@@ -14,12 +14,12 @@ class BaseConfig(AppConfig):
name = "base" name = "base"
def ready(self) -> None: def ready(self) -> None:
ready = super().ready() super().ready()
try: try:
from base.models import EmployeeShiftDay from base.models import EmployeeShiftDay
if len(EmployeeShiftDay.objects.all()) == 0: if not EmployeeShiftDay.objects.exists():
days = ( days = [
("monday", "Monday"), ("monday", "Monday"),
("tuesday", "Tuesday"), ("tuesday", "Tuesday"),
("wednesday", "Wednesday"), ("wednesday", "Wednesday"),
@@ -27,11 +27,10 @@ class BaseConfig(AppConfig):
("friday", "Friday"), ("friday", "Friday"),
("saturday", "Saturday"), ("saturday", "Saturday"),
("sunday", "Sunday"), ("sunday", "Sunday"),
]
EmployeeShiftDay.objects.bulk_create(
[EmployeeShiftDay(day=day[0]) for day in days]
) )
for day in days:
shift_day = EmployeeShiftDay()
shift_day.day = day[0]
shift_day.save()
except Exception as e: except Exception as e:
pass pass
return ready

View File

@@ -1,3 +1,7 @@
"""
Horilla management command to create a new user and associated employee.
"""
import uuid import uuid
from django.contrib.auth.models import User from django.contrib.auth.models import User
@@ -7,6 +11,10 @@ from employee.models import Employee
class Command(BaseCommand): class Command(BaseCommand):
"""
Horilla management command to create a new user and associated employee.
"""
help = "Creates a new user" help = "Creates a new user"
def add_arguments(self, parser): def add_arguments(self, parser):
@@ -33,31 +41,35 @@ class Command(BaseCommand):
email = options["email"] email = options["email"]
phone = options["phone"] phone = options["phone"]
adminuser = User.objects.filter(username=first_name).first() if User.objects.filter(username=username).exists():
if adminuser is not None: self.stdout.write(
self.stdout.write(self.style.WARNING('User "%s" already exist' % adminuser)) self.style.WARNING(f'User with username "{username}" already exists')
else: )
try: return
user = User.objects.create_superuser(
username=username, email=email, password=password
)
employee = Employee()
employee.employee_user_id = user
employee.employee_first_name = first_name
employee.employee_last_name = last_name
employee.email = email
employee.phone = phone
employee.save()
bot = User.objects.filter(username="Horilla Bot").first()
if bot is None:
User.objects.create_user(
username="Horilla Bot",
password=str(uuid.uuid4()),
)
self.stdout.write( try:
self.style.SUCCESS('Employee "%s" created successfully' % employee) user = User.objects.create_superuser(
username=username, email=email, password=password
)
employee = Employee()
employee.employee_user_id = user
employee.employee_first_name = first_name
employee.employee_last_name = last_name
employee.email = email
employee.phone = phone
employee.save()
bot = User.objects.filter(username="Horilla Bot").first()
if bot is None:
User.objects.create_user(
username="Horilla Bot",
password=str(uuid.uuid4()),
) )
except Exception as e:
self.stdout.write(
self.style.SUCCESS(f'Employee "{employee}" created successfully')
)
except Exception as e:
if "user" in locals():
user.delete() user.delete()
raise CommandError('Error creating user "%s": %s' % (username, e)) raise CommandError(f'Error creating user "{username}": {e}') from e

View File

@@ -4,12 +4,11 @@ middleware.py
from django.apps import apps from django.apps import apps
from django.db.models import Q from django.db.models import Q
from django.http import HttpResponse, HttpResponseNotAllowed from django.shortcuts import redirect
from django.shortcuts import redirect, render
from base.context_processors import AllCompany from base.context_processors import AllCompany
from base.horilla_company_manager import HorillaCompanyManager from base.horilla_company_manager import HorillaCompanyManager
from base.models import Company, ShiftRequest, WorkTypeRequest from base.models import ShiftRequest, WorkTypeRequest
from employee.models import ( from employee.models import (
DisciplinaryAction, DisciplinaryAction,
Employee, Employee,
@@ -23,149 +22,156 @@ from horilla_documents.models import DocumentRequest
class CompanyMiddleware: class CompanyMiddleware:
""" """
company middleware class Middleware to handle company-specific filtering for models.
"""
def __init__(self, get_response):
self.get_response = get_response
def _get_company_id(self, request):
"""
Retrieve the company ID from the request or session.
"""
if getattr(request, "user", False) and not request.user.is_anonymous:
try:
return getattr(
request.user.employee_get.employee_work_info, "company_id", None
)
except AttributeError:
pass
return None
def _set_company_session(self, request, company_id):
"""
Set the company session data based on the company ID.
"""
if company_id and request.session.get("selected_company") != "all":
request.session["selected_company"] = company_id.id
request.session["selected_company_instance"] = {
"company": company_id.company,
"icon": company_id.icon.url,
"text": "My company",
"id": company_id.id,
}
else:
request.session["selected_company"] = "all"
all_company = AllCompany()
request.session["selected_company_instance"] = {
"company": all_company.company,
"icon": all_company.icon.url,
"text": all_company.text,
"id": all_company.id,
}
def _add_company_filter(self, model, company_id):
"""
Add company filter to the model if applicable.
"""
is_company_model = model in self._get_company_models()
company_field = getattr(model, "company_id", None)
is_horilla_manager = isinstance(model.objects, HorillaCompanyManager)
related_company_field = getattr(model.objects, "related_company_field", None)
if is_company_model:
if company_field:
model.add_to_class("company_filter", Q(company_id=company_id))
elif is_horilla_manager and related_company_field:
model.add_to_class(
"company_filter", Q(**{related_company_field: company_id})
)
else:
if company_field:
model.add_to_class(
"company_filter",
Q(company_id=company_id) | Q(company_id__isnull=True),
)
elif is_horilla_manager and related_company_field:
model.add_to_class(
"company_filter",
Q(**{related_company_field: company_id})
| Q(**{f"{related_company_field}__isnull": True}),
)
def _get_company_models(self):
"""
Retrieve the list of models that are company-specific.
"""
company_models = [
Employee,
ShiftRequest,
WorkTypeRequest,
DocumentRequest,
DisciplinaryAction,
EmployeeBankDetails,
EmployeeWorkInformation,
]
app_model_mappings = {
"recruitment": ["recruitment", "candidate"],
"leave": [
"leaverequest",
"restrictleave",
"availableleave",
"leaveallocationrequest",
"compensatoryleaverequest",
],
"asset": ["assetassignment", "assetrequest"],
"attendance": [
"attendance",
"attendanceactivity",
"attendanceovertime",
"workrecords",
],
"payroll": [
"contract",
"loanaccount",
"payslip",
"reimbursement",
"workrecord",
],
"helpdesk": ["ticket"],
"offboarding": ["offboarding"],
"pms": ["employeeobjective"],
}
for app_label, models in app_model_mappings.items():
if apps.is_installed(app_label):
company_models.extend(
[get_horilla_model_class(app_label, model) for model in models]
)
return company_models
def __call__(self, request):
if getattr(request, "user", False) and not request.user.is_anonymous:
company_id = self._get_company_id(request)
self._set_company_session(request, company_id)
app_models = [
model for model in apps.get_models() if model._meta.app_label in APPS
]
for model in app_models:
self._add_company_filter(model, company_id)
response = self.get_response(request)
return response
class ForcePasswordChangeMiddleware:
"""
Middleware to force password change for new employees.
""" """
def __init__(self, get_response): def __init__(self, get_response):
self.get_response = get_response self.get_response = get_response
def __call__(self, request): def __call__(self, request):
# Get the current user's company_id from the request
if getattr(request, "user", False) and not request.user.is_anonymous:
company_id = None
try:
company_id = getattr(
request.user.employee_get.employee_work_info, "company_id", None
)
except:
pass
if (
request.session.get("selected_company")
and request.session.get("selected_company") != "all"
):
company_id = Company.objects.filter(
id=request.session.get("selected_company")
).first()
elif company_id and request.session.get("selected_company") != "all":
request.session["selected_company"] = company_id.id
request.session["selected_company_instance"] = {
"company": company_id.company,
"icon": company_id.icon.url,
"text": "My company",
"id": company_id.id,
}
elif not company_id:
request.session["selected_company"] = "all"
all_company = AllCompany()
request.session["selected_company_instance"] = {
"company": all_company.company,
"icon": all_company.icon.url,
"text": all_company.text,
"id": all_company.id,
}
app_models = [
model for model in apps.get_models() if model._meta.app_label in APPS
]
company_models = [
Employee,
ShiftRequest,
WorkTypeRequest,
DocumentRequest,
DisciplinaryAction,
EmployeeBankDetails,
EmployeeWorkInformation,
]
app_model_mappings = {
"recruitment": ["recruitment", "candidate"],
"leave": [
"leaverequest",
"restrictleave",
"availableleave",
"leaveallocationrequest",
"compensatoryleaverequest",
],
"asset": ["assetassignment", "assetrequest"],
"attendance": [
"attendance",
"attendanceactivity",
"attendanceovertime",
"workrecords",
],
"payroll": [
"contract",
"loanaccount",
"payslip",
"reimbursement",
"workrecord",
],
"helpdesk": ["ticket"],
"offboarding": ["offboarding"],
"pms": ["employeeobjective"],
}
# Dynamically add models if their respective apps are installed
for app_label, models in app_model_mappings.items():
if apps.is_installed(app_label):
company_models.extend(
[get_horilla_model_class(app_label, model) for model in models]
)
if company_id:
for model in app_models:
is_company_model = model in company_models
company_field = getattr(model, "company_id", None)
is_horilla_manager = isinstance(
model.objects, HorillaCompanyManager
)
related_company_field = getattr(
model.objects, "related_company_field", None
)
if is_company_model:
if company_field:
model.add_to_class(
"company_filter", Q(company_id=company_id)
)
elif is_horilla_manager and related_company_field:
model.add_to_class(
"company_filter",
Q(**{related_company_field: company_id}),
)
else:
if company_field:
model.add_to_class(
"company_filter",
Q(company_id=company_id) | Q(company_id__isnull=True),
)
elif is_horilla_manager and related_company_field:
model.add_to_class(
"company_filter",
Q(**{related_company_field: company_id})
| Q(**{f"{related_company_field}__isnull": True}),
)
response = self.get_response(request)
return response
# MIDDLEWARE TO CHECK IF EMPLOYEE IS NEW USER OR NOT
class ForcePasswordChangeMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Exclude specific paths from redirection
excluded_paths = ["/change-password", "/login", "/logout"] excluded_paths = ["/change-password", "/login", "/logout"]
if request.path.rstrip("/") in excluded_paths: if request.path.rstrip("/") in excluded_paths:
return self.get_response(request) return self.get_response(request)
# Check if employee is a new employee
if hasattr(request, "user") and request.user.is_authenticated: if hasattr(request, "user") and request.user.is_authenticated:
if getattr(request.user, "is_new_employee", True): if getattr(request.user, "is_new_employee", True):
return redirect("change-password") # Adjust to match your URL name return redirect("change-password")
return self.get_response(request) return self.get_response(request)

View File

@@ -24,7 +24,6 @@ from django.contrib.auth.views import PasswordResetConfirmView, PasswordResetVie
from django.core.files.base import ContentFile from django.core.files.base import ContentFile
from django.core.mail import EmailMultiAlternatives from django.core.mail import EmailMultiAlternatives
from django.core.management import call_command from django.core.management import call_command
from django.core.paginator import Paginator
from django.db.models import ProtectedError, Q from django.db.models import ProtectedError, Q
from django.http import Http404, HttpResponse, HttpResponseRedirect, JsonResponse from django.http import Http404, HttpResponse, HttpResponseRedirect, JsonResponse
from django.shortcuts import get_object_or_404, redirect, render from django.shortcuts import get_object_or_404, redirect, render
@@ -32,11 +31,11 @@ from django.template.loader import render_to_string
from django.urls import reverse, reverse_lazy from django.urls import reverse, reverse_lazy
from django.utils import timezone from django.utils import timezone
from django.utils.html import strip_tags from django.utils.html import strip_tags
from django.utils.timezone import localdate
from django.utils.translation import gettext as _ from django.utils.translation import gettext as _
from django.views.decorators.csrf import csrf_exempt from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods from django.views.decorators.http import require_http_methods
from accessibility.accessibility import ACCESSBILITY_FEATURE
from accessibility.models import DefaultAccessibility from accessibility.models import DefaultAccessibility
from base.backends import ConfiguredEmailBackend from base.backends import ConfiguredEmailBackend
from base.decorators import ( from base.decorators import (
@@ -112,18 +111,14 @@ from base.methods import (
format_date, format_date,
generate_colors, generate_colors,
get_key_instances, get_key_instances,
get_pagination,
is_reportingmanager, is_reportingmanager,
paginator_qry, paginator_qry,
random_color_generator,
sortby, sortby,
) )
from base.models import ( from base.models import (
WEEK_DAYS, WEEK_DAYS,
WEEKS, WEEKS,
Announcement,
AnnouncementExpire, AnnouncementExpire,
AnnouncementView,
BaserequestFile, BaserequestFile,
BiometricAttendance, BiometricAttendance,
Company, Company,
@@ -133,7 +128,6 @@ from base.models import (
DynamicEmailConfiguration, DynamicEmailConfiguration,
DynamicPagination, DynamicPagination,
EmployeeShift, EmployeeShift,
EmployeeShiftDay,
EmployeeShiftSchedule, EmployeeShiftSchedule,
EmployeeType, EmployeeType,
Holidays, Holidays,
@@ -220,15 +214,15 @@ def initialize_database_condition():
Returns: Returns:
bool: True if the database needs to be initialized, False otherwise. bool: True if the database needs to be initialized, False otherwise.
""" """
initialize_database = not User.objects.exists() init_database = not User.objects.exists()
if not initialize_database: if not init_database:
initialize_database = True init_database = True
superusers = User.objects.filter(is_superuser=True) superusers = User.objects.filter(is_superuser=True)
for user in superusers: for user in superusers:
if hasattr(user, "employee_get"): if hasattr(user, "employee_get"):
initialize_database = False init_database = False
break break
return initialize_database return init_database
def load_demo_database(request): def load_demo_database(request):
@@ -284,9 +278,7 @@ def initialize_database(request):
if initialize_database_condition(): if initialize_database_condition():
if request.method == "POST": if request.method == "POST":
password = request._post.get("password") password = request._post.get("password")
from horilla.horilla_settings import DB_INIT_PASSWORD as db_password if DB_INIT_PASSWORD == password:
if db_password == password:
return redirect(initialize_database_user) return redirect(initialize_database_user)
else: else:
messages.warning( messages.warning(
@@ -1313,9 +1305,11 @@ def add_remove_dynamic_fields(request, **kwargs):
- model (Model): The Django model class used for `ModelChoiceField`. - model (Model): The Django model class used for `ModelChoiceField`.
- form_class (Form): The Django form class to which dynamic fields will be added. - form_class (Form): The Django form class to which dynamic fields will be added.
- template (str): The template used to render the newly added field. - template (str): The template used to render the newly added field.
- empty_label (str, optional): The label to show for empty choices in a `ModelChoiceField`. - empty_label (str, optional): The label to show for empty choices in
a `ModelChoiceField`.
- field_name_pre (str): The prefix for the dynamically generated field names. - field_name_pre (str): The prefix for the dynamically generated field names.
- field_type (str, optional): The type of field to add, either "character" or "model_choice". - field_type (str, optional): The type of field to add, either "character"
or "model_choice".
Returns: Returns:
HttpResponse: Returns the HTML for the newly added field, rendered in the context of the HttpResponse: Returns the HTML for the newly added field, rendered in the context of the
@@ -5199,9 +5193,6 @@ def enable_account_block_unblock(request):
return HttpResponse(status=405) return HttpResponse(status=405)
from accessibility.accessibility import ACCESSBILITY_FEATURE
@login_required @login_required
@permission_required("employee.change_employee") @permission_required("employee.change_employee")
def enable_profile_edit_feature(request): def enable_profile_edit_feature(request):
@@ -6579,7 +6570,7 @@ def get_upcoming_holidays(request):
""" """
Retrieve and display a list of upcoming holidays for the current month and year. Retrieve and display a list of upcoming holidays for the current month and year.
""" """
today = localdate() # This accounts for timezone-aware dates today = timezone.localdate()
current_month = today.month current_month = today.month
current_year = today.year current_year = today.year
holidays = Holidays.objects.filter( holidays = Holidays.objects.filter(

View File

@@ -306,7 +306,6 @@ class HorillaListView(ListView):
context["keys_to_remove"] = keys_to_remove context["keys_to_remove"] = keys_to_remove
request = self.request request = self.request
ordered_ids = list(queryset.values_list("id", flat=True))
is_first_sort = False is_first_sort = False
query_dict = self.request.GET query_dict = self.request.GET
if ( if (