Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 74 additions & 0 deletions docs/domains/shifts.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
# Shifts

## Domain

### Models

- `BusStop`: defines arrival + departure time from a `Place`
- `BusShift`: driver and bus assigned to a list of `BusStop`

Possible improvements (depending on business logic):
- Make `BusStop` a true JOIN table connecting `Place` and `BusShift`
if a `BusStop` can only belong to one `BusShift`.
- Remove `date` logic and keep only `time` if a `BusShift` is expected
to run the same on different day (could also add logic differentiating
weekdays/weekends/holidays/etc.)
- Add check that `arrival_time`/`departure_time` on `BusStop` objects
is coherent with average bus speed and location of stops.

### Signals

- `create_shift_managers_group`: creates Shift managers group
and assigns `can_manage_shifts` permission. Runs when migrations
for `shifts` app have run. Caveat: will run after each new migration.
Should be updated to run only once after relevant migration has gone
through. Perhaps a separate data migration would be a better choice.
- `validate_bus_shift_stops`: uses `validate_stops` validator
when updating/deleting/clearing a `BusShift`. Currently subject to
a possible TOCTOU race condition. Could be fixed with a `select_for_update()`
and `transaction.atomic()` in the form save path, but won't be effective
on SQLite, so did not implement.

### Admin

* Separate admin panel allowing access to `Shift managers` group (RBAC logic) only to
`BusStop` and `BusShift` objects with permission to edit/add, but not
delete (sensitive action). Can be updated depending on business logic.
* Main admin panel could let `staff` users access the objects if
users were granted the auto-created permission on the model (`view_busshift`)
outside of the group. Not particularly problematic for this test.
* In a real-life case, an actual front-end interface would be better
suited to allowing users to perform this task, hence the choice to
separate the admin sites. The form aspect to modify `BusStop` objects
from a `BusShift` could be improved.

### Validators

* `validate_stops` is used both by the signal and the admin panel
form to have a single source of truth. Using `Min/Max` annotations
to avoid running N+1 queries. `ValidationError` in the validator applies to
admin panel whilst signal catches it and raises `IntegrityError` for
DB-level.

### Commands

* `create_shifts`: creates bus shifts with associated stops

Improvements:
* Add command to create user in `ShiftManager` group.

## Run

Access the shift admin panel in local at `http://localhost:8000/shift-admin`.
You need a `superuser` or `shift manager` account to connect.

Run tests:
```commandline
python3 manage.py tests padam_django.apps.shifts
```

Create `N` shifts (with `N^2` stops,
`N` of those randomly assigned to each shift):
```commandline
python3 manage.py create_shifts -n N
```
Original file line number Diff line number Diff line change
Expand Up @@ -12,3 +12,4 @@ def handle(self, *args, **options):
management.call_command('create_drivers', number=5)
management.call_command('create_buses', number=10)
management.call_command('create_places', number=30)
management.call_command('create_shifts', number=5)
Empty file.
120 changes: 120 additions & 0 deletions padam_django/apps/shifts/admin.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
from django.contrib import admin
from django.contrib.admin.forms import AuthenticationForm
from django.db.models import Max, Min
from django.db.models.query import QuerySet
from django.http.request import HttpRequest

from . import models
from .forms import BusShiftForm


class ShiftAdminSite(admin.AdminSite):
site_header = "Shift Admin Portal"
site_title = "Shift Admin"
index_title = "Shift management"
login_form = AuthenticationForm

def has_permission(self, request: HttpRequest) -> bool:
return request.user.is_authenticated and (
request.user.is_superuser
or request.user.groups.filter(name="Shift managers").exists()
)


shift_admin = ShiftAdminSite(name="shift_admin")


@admin.register(models.BusStop)
class BusStopAdmin(admin.ModelAdmin):
list_display = ["id", "get_place_name", "arrival_time", "departure_time"]

def get_place_name(self, obj: models.BusStop) -> str:
return obj.place.name

get_place_name.short_description = "Place"

def has_change_permission(
self, request: HttpRequest, obj: models.BusShift | None = None
) -> bool:
if request.user.has_perm("shifts.can_manage_shifts"):
return True
return False

def has_delete_permission(
self, request: HttpRequest, obj: models.BusShift | None = None
) -> bool:
return False

def has_add_permission(
self, request: HttpRequest, obj: models.BusShift | None = None
) -> bool:
if request.user.has_perm("shifts.can_manage_shifts"):
return True
return False


@admin.register(models.BusShift)
class BusShiftAdmin(admin.ModelAdmin):
list_display = [
"id",
"get_bus_licence_plate",
"get_driver_username",
"get_start_time",
"get_end_time",
]
list_filter = ["bus__licence_plate"]
search_fields = ["bus__licence_plate", "driver__user__username"]
form = BusShiftForm
list_select_related = ["bus", "driver__user"]

def get_queryset(self, request: HttpRequest) -> QuerySet[models.BusShift]:
return (
super()
.get_queryset(request)
.annotate(
_start_time=Min("stops__departure_time"),
_end_time=Max("stops__arrival_time"),
)
)

def get_bus_licence_plate(self, obj: models.BusShift) -> str:
return obj.bus.licence_plate

def get_driver_username(self, obj: models.BusShift) -> str:
return obj.driver.user.username

def get_start_time(self, obj: models.BusShift):
return obj._start_time

def get_end_time(self, obj: models.BusShift):
return obj._end_time

get_bus_licence_plate.short_description = "Bus"
get_driver_username.short_description = "Driver"
get_start_time.short_description = "Start time"
get_start_time.admin_order_field = "_start_time"
get_end_time.short_description = "End time"
get_end_time.admin_order_field = "_end_time"

def has_change_permission(
self, request: HttpRequest, obj: models.BusShift | None = None
) -> bool:
if request.user.has_perm("shifts.can_manage_shifts"):
return True
return False

def has_delete_permission(
self, request: HttpRequest, obj: models.BusShift | None = None
) -> bool:
return False

def has_add_permission(
self, request: HttpRequest, obj: models.BusShift | None = None
) -> bool:
if request.user.has_perm("shifts.can_manage_shifts"):
return True
return False


shift_admin.register(models.BusStop, BusStopAdmin)
shift_admin.register(models.BusShift, BusShiftAdmin)
10 changes: 10 additions & 0 deletions padam_django/apps/shifts/apps.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
from django.apps import AppConfig


class ShiftsConfig(AppConfig):
name = "padam_django.apps.shifts"
models_module = "padam_django.apps.shifts"
label = "shifts"

def ready(self):
import padam_django.apps.shifts.signals # noqa: F401
37 changes: 37 additions & 0 deletions padam_django/apps/shifts/factories.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
import datetime

import factory
from django.utils.timezone import now
from faker import Faker

from . import models

fake = Faker(["fr"])


class BusStopFactory(factory.django.DjangoModelFactory):
place = factory.SubFactory("padam_django.apps.geography.factories.PlaceFactory")
arrival_time = factory.Faker(
"date_time_between", start_date=now().date(), tzinfo=datetime.timezone.utc
)
departure_time = factory.LazyAttribute(
lambda o: o.arrival_time
+ datetime.timedelta(minutes=fake.random_int(min=1, max=3))
)

class Meta:
model = models.BusStop


class BusShiftFactory(factory.django.DjangoModelFactory):
driver = factory.SubFactory("padam_django.apps.fleet.factories.DriverFactory")
bus = factory.SubFactory("padam_django.apps.fleet.factories.BusFactory")

@factory.post_generation
def stops(self, create: bool, extracted: list, **kwargs: dict):
if not create or not extracted:
return
self.stops.add(*extracted)

class Meta:
model = models.BusShift
28 changes: 28 additions & 0 deletions padam_django/apps/shifts/forms.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
from typing import Any

from django import forms
from django.core.exceptions import ValidationError

from padam_django.apps.shifts.models import BusShift
from padam_django.apps.shifts.validators import validate_stops


class BusShiftForm(forms.ModelForm):
class Meta:
model = BusShift
fields = "__all__"

def clean(self) -> dict[str, Any] | None:
cleaned_data = super().clean()
if not cleaned_data:
return None
stops = cleaned_data.get("stops")
driver = cleaned_data.get("driver")
bus = cleaned_data.get("bus")

if stops and driver and bus:
try:
validate_stops(driver, bus, list(stops), exclude_pk=self.instance.pk)
except ValidationError as e:
raise forms.ValidationError(e.message)
return cleaned_data
Empty file.
Empty file.
19 changes: 19 additions & 0 deletions padam_django/apps/shifts/management/commands/create_shifts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
import random

from padam_django.apps.common.management.base import CreateDataBaseCommand
from padam_django.apps.shifts.factories import BusShiftFactory, BusStopFactory


class Command(CreateDataBaseCommand):

help = "Create a few shifts"

def handle(self, *args, **options):
super().handle(*args, **options)
self.stdout.write(
f"Creating {self.number} shifts and {self.number ** 2} stops ..."
)
stops = BusStopFactory.create_batch(size=self.number**2)
# Doing this to have different stops, not ideal
for ind in range(self.number):
BusShiftFactory.create(stops=random.sample(stops, self.number))
90 changes: 90 additions & 0 deletions padam_django/apps/shifts/migrations/0001_initial.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
# Generated by Django 4.2.16 on 2026-08-04 17:25

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


class Migration(migrations.Migration):

initial = True

dependencies = [
("geography", "0001_initial"),
("fleet", "0002_auto_20211109_1456"),
]

operations = [
migrations.CreateModel(
name="BusStop",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("arrival_time", models.DateTimeField()),
("departure_time", models.DateTimeField()),
(
"place",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="bus_stop",
to="geography.place",
),
),
],
options={
"db_table": "bus_stop",
"ordering": ["arrival_time"],
},
),
migrations.CreateModel(
name="BusShift",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"bus",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="bus_shift",
to="fleet.bus",
),
),
(
"driver",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="bus_shift",
to="fleet.driver",
),
),
("stops", models.ManyToManyField(to="shifts.busstop")),
],
options={
"db_table": "bus_shift",
},
),
migrations.AddConstraint(
model_name="busstop",
constraint=models.CheckConstraint(
check=models.Q(("arrival_time__lt", models.F("departure_time"))),
name="bus_stop_arrival_time__lt_departure_time",
),
),
migrations.AlterModelOptions(
name="busshift",
options={"permissions": (("can_manage_shifts", "Can manage shifts"),)},
),
]
Empty file.
Loading