This commit is contained in:
Ross
2025-12-13 22:41:32 +00:00
parent d509922450
commit 249e09766d
10 changed files with 296 additions and 11 deletions
@@ -0,0 +1,18 @@
# Generated by assistant for adding nullable request_token
from django.db import migrations, models
import uuid
class Migration(migrations.Migration):
dependencies = [
("rota", "0004_worker_options"),
]
operations = [
migrations.AddField(
model_name="worker",
name="request_token",
field=models.UUIDField(default=uuid.uuid4, editable=False, null=True),
),
]
@@ -0,0 +1,26 @@
from django.db import migrations
import uuid
def generate_tokens(apps, schema_editor):
Worker = apps.get_model('rota', 'Worker')
db_alias = schema_editor.connection.alias
for w in Worker.objects.using(db_alias).all():
if not getattr(w, 'request_token', None):
new_token = uuid.uuid4()
# extremely unlikely collision, but guard anyway
while Worker.objects.using(db_alias).filter(request_token=new_token).exists():
new_token = uuid.uuid4()
w.request_token = new_token
w.save(update_fields=['request_token'])
class Migration(migrations.Migration):
dependencies = [
('rota', '0005_add_request_token'),
]
operations = [
migrations.RunPython(generate_tokens, reverse_code=migrations.RunPython.noop),
]
@@ -0,0 +1,16 @@
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('rota', '0006_populate_request_token'),
]
operations = [
migrations.AlterField(
model_name='worker',
name='request_token',
field=models.UUIDField(editable=False, unique=True),
),
]
+9
View File
@@ -1,5 +1,6 @@
from django.db import models
from django.utils import timezone
import uuid
class RotaSchedule(models.Model):
@@ -130,6 +131,14 @@ class Worker(models.Model):
rotas = models.ManyToManyField(RotaSchedule, through="Assignment", related_name="workers")
# Token used to allow workers (without Django accounts) to access
# self-service links (request leave, view their calendar, etc.). This
# is a UUID4 and should be treated as a secret. Regenerate to revoke.
# During migration we add this field as nullable and non-unique first,
# then populate unique values in a data migration and finally make it
# non-null and unique. See migrations/ for the migration sequence.
request_token = models.UUIDField(default=uuid.uuid4, editable=False, null=True, unique=False)
def __str__(self):
return self.name
+2
View File
@@ -17,6 +17,8 @@ urlpatterns = [
path("worker/<int:rota_id>/<int:worker_id>/delete/", views.worker_delete, name="worker_delete"),
path("worker/<int:worker_id>/", views.worker_detail, name="worker_detail"),
path("leave/request/", views.request_leave, name="request_leave"),
path("leave/request/token/<uuid:token>/", views.request_leave, name="request_leave_token"),
path("worker/<int:worker_id>/regenerate_token/", views.regenerate_worker_token, name="regenerate_worker_token"),
path("run/<int:run_id>/", views.rota_run_detail, name="rota_run_detail"),
path("run/<int:run_id>/export/html/", views.rota_run_export_html, name="rota_run_export_html"),
path("run/<int:run_id>/export/download/", views.rota_run_export_download, name="rota_run_export_download"),
+57 -10
View File
@@ -304,7 +304,18 @@ def worker_detail(request, worker_id):
leave.save()
return redirect("rota:worker_detail", worker_id=worker.id)
else:
form = LeaveForm()
# Pre-fill dates when `date` query param is provided (ISO YYYY-MM-DD)
date_q = request.GET.get("date") or request.POST.get("date")
if date_q:
try:
import datetime
d = datetime.date.fromisoformat(date_q)
form = LeaveForm(initial={"start_date": d, "end_date": d})
except Exception:
form = LeaveForm()
else:
form = LeaveForm()
return render(request, "rota/worker_detail.html", {"worker": worker, "form": form})
@@ -340,12 +351,40 @@ def worker_edit(request, rota_id, worker_id):
form = WorkerForm(instance=worker)
modal_html = render_to_string(
"rota/partials/worker_form_modal.html",
{"form": form, "form_action": request.path, "rota": rota},
{"form": form, "form_action": request.path, "rota": rota, "worker": worker, "token_link": request.build_absolute_uri(reverse('rota:request_leave_token', args=[worker.request_token])) if worker and getattr(worker, 'request_token', None) else ''},
request=request,
)
return HttpResponse(modal_html)
@require_POST
def regenerate_worker_token(request, worker_id):
# Only allow staff users to regenerate tokens from the admin modal
if not (request.user and request.user.is_authenticated and request.user.is_staff):
return HttpResponseBadRequest("Permission denied")
worker = get_object_or_404(Worker, pk=worker_id)
# generate unique token
import uuid
new_token = uuid.uuid4()
# guard against collision
while Worker.objects.filter(request_token=new_token).exclude(pk=worker.pk).exists():
new_token = uuid.uuid4()
worker.request_token = new_token
worker.save(update_fields=["request_token"])
# render only the token area so HTMX can swap it
token_html = render_to_string(
"rota/partials/worker_token_area.html",
{"worker": worker, "token_link": request.build_absolute_uri(reverse('rota:request_leave_token', args=[worker.request_token])) if getattr(worker, 'request_token', None) else ''},
request=request,
)
response = HttpResponse(token_html)
response["HX-Trigger"] = "workerTokenRegenerated"
return response
@require_http_methods(["GET", "POST"])
def worker_delete(request, rota_id, worker_id):
rota = get_object_or_404(RotaSchedule, pk=rota_id)
@@ -421,7 +460,7 @@ def rota_options_update(request, rota_id):
@require_http_methods(["GET", "POST"])
def request_leave(request):
def request_leave(request, token=None):
"""Allow a worker to request leave for themselves.
Resolution strategy:
@@ -434,14 +473,22 @@ def request_leave(request):
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
worker = None
# try authenticated user -> worker by email
try:
if request.user and request.user.is_authenticated and getattr(request.user, "email", None):
worker = Worker.objects.filter(email=request.user.email).first()
except Exception:
worker = None
# 1) token link (preferred for unauthenticated users)
if token:
try:
worker = get_object_or_404(Worker, request_token=token)
except Exception:
worker = None
# fallback to explicit worker_id
# 2) try authenticated user -> worker by email
if worker is None:
try:
if request.user and request.user.is_authenticated and getattr(request.user, "email", None):
worker = Worker.objects.filter(email=request.user.email).first()
except Exception:
worker = None
# 3) fallback to explicit worker_id param
if worker is None:
worker_id = request.GET.get("worker_id") or request.POST.get("worker_id")
if worker_id: