-
Notifications
You must be signed in to change notification settings - Fork 244
/
Copy paththreading.py
172 lines (137 loc) · 7.19 KB
/
threading.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import logging
from threading import Thread
from django.contrib import messages
from django.core.mail import EmailMessage
from django.db.models import Q
from django.template.loader import render_to_string
from django.utils.translation import gettext as _
from base.backends import ConfiguredEmailBackend
logger = logging.getLogger(__name__)
class LeaveMailSendThread(Thread):
def __init__(self, request, leave_request, type):
Thread.__init__(self)
self.request = request
self.leave_request = leave_request
self.type = type
self.host = request.get_host()
self.protocol = "https" if request.is_secure() else "http"
def send_email(self, subject, content, recipients, leave_request_id="#"):
email_backend = ConfiguredEmailBackend()
display_email_name = email_backend.dynamic_from_email_with_display_name
if self.request:
try:
display_email_name = f"{self.request.user.employee_get.get_full_name()} <{self.request.user.employee_get.email}>"
except:
logger.error(Exception)
host = self.host
protocol = self.protocol
if leave_request_id != "#":
link = int(leave_request_id)
for recipient in recipients:
if recipient:
html_message = render_to_string(
"base/mail_templates/leave_request_template.html",
{
"link": link,
"instance": recipient,
"host": host,
"protocol": protocol,
"subject": subject,
"content": content,
},
request=self.request,
)
email = EmailMessage(
subject=subject,
body=html_message,
from_email=display_email_name,
to=[recipient.email],
reply_to=[display_email_name],
)
email.content_subtype = "html"
try:
email.send()
except:
messages.error(
self.request, f"Mail not sent to {recipient.get_full_name()}"
)
def run(self) -> None:
super().run()
if self.type == "request":
owner = self.leave_request.employee_id
reporting_manager = self.leave_request.employee_id.get_reporting_manager()
content_manager = f"This is to inform you that a leave request has been requested by {owner}. Take the necessary actions for the leave request. Should you have any additional information or updates, please feel free to communicate directly with the {owner}."
subject_manager = f"Leave request has been requested by {owner}"
self.send_email(
subject_manager,
content_manager,
[reporting_manager],
self.leave_request.id,
)
content_owner = f"This is to inform you that the leave request you created has been successfully logged into our system. The manager will now take the necessary actions to address leave request. Should you have any additional information or updates, please feel free to communicate directly with the {reporting_manager}."
subject_owner = "Leave request created successfully"
self.send_email(
subject_owner, content_owner, [owner], self.leave_request.id
)
elif self.type == "approve":
owner = self.leave_request.employee_id
reporting_manager = self.leave_request.employee_id.get_reporting_manager()
subject = "The Leave request has been successfully approved"
content = f"This is to inform you that the leave request has been approved. If you have any questions or require further information, feel free to reach out to the {reporting_manager}."
self.send_email(subject, content, [owner], self.leave_request.id)
elif self.type == "reject":
owner = self.leave_request.employee_id
reporting_manager = self.leave_request.employee_id.get_reporting_manager()
subject = "The Leave request has been rejected"
content = f"This is to inform you that the leave request has been rejected. If you have any questions or require further information, feel free to reach out to the {reporting_manager}."
self.send_email(subject, content, [owner], self.leave_request.id)
elif self.type == "cancel":
owner = self.leave_request.employee_id
reporting_manager = self.leave_request.employee_id.get_reporting_manager()
content_manager = f"This is to inform you that a leave request has been requested to cancel by {owner}. Take the necessary actions for the leave request. Should you have any additional information or updates, please feel free to communicate directly with the {owner}."
subject_manager = f"Leave request cancellation"
self.send_email(
subject_manager,
content_manager,
[reporting_manager],
self.leave_request.id,
)
content_owner = f"This is to inform you that a cancellation request created for your leave request has been successfully logged into our system. The manager will now take the necessary actions to address the leave request. Should you have any additional information or updates, please feel free to communicate directly with the {reporting_manager}."
subject_owner = "Leave request cancellation requested"
self.send_email(
subject_owner, content_owner, [owner], self.leave_request.id
)
return
class LeaveClashThread(Thread):
def __init__(self, leave_request):
Thread.__init__(self)
self.leave_request = leave_request
def count_leave_clashes(self):
from leave.models import LeaveRequest
"""
Method to count leave clashes where this employee's leave request overlaps
with other employees' requested dates.
"""
overlapping_requests = LeaveRequest.objects.exclude(
id=self.leave_request.id
).filter(
Q(
employee_id__employee_work_info__department_id=self.leave_request.employee_id.employee_work_info.department_id
)
| Q(
employee_id__employee_work_info__job_position_id=self.leave_request.employee_id.employee_work_info.job_position_id
),
start_date__lte=self.leave_request.end_date,
end_date__gte=self.leave_request.start_date,
)
return overlapping_requests.count()
def run(self) -> None:
from leave.models import LeaveRequest
super().run()
dates = self.leave_request.requested_dates()
leave_requests_to_update = LeaveRequest.objects.filter(
Q(start_date__in=dates) | Q(end_date__in=dates)
)
for leave_request in leave_requests_to_update:
leave_request.leave_clashes_count = self.count_leave_clashes()
leave_request.save()