gitea
4b6138f049
Four related backend features. They are committed together because they interleave in api/views.py and api/serializers.py - one hunk holds the combinations ViewSet, the licensing helpers and LicenseView - and splitting by hunk would produce commits that do not import. ExtraCombination records previously-used combinations of lookup nomenclature values so an operator can refill the Additional Data fields in one click. Deduplicated by a canonical order-independent hash, since JSONField equality compares serialized text on SQLite and so depends on key order. Weighing replaces set_tare/set_gross with a single measure action that assigns by weight - lighter becomes tare, heavier gross - because trucks arrive both loaded and empty. A second, lighter reading moves the first to gross carrying its own timestamp, user and manual fields, so provenance follows the value rather than the slot. Manual entries record who, when and why, gated on the previously unenforced can_measure and can_manually_measure flags. ScaleProtocol stores how to read each indicator, so a new scale type is a configuration change rather than a new binary on every station PC. Licensing is offline and node-locked: the server verifies an Ed25519 signature against a fingerprint of four hardware components, tolerating one change, and counts concurrently connected stations. Sites are air-gapped, so there is no activation server. A 14-day grace period keeps an unlicensed or newly-invalid install working, because a licence check must never stop a weighbridge without warning. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
636 lines
24 KiB
Python
636 lines
24 KiB
Python
from rest_framework import viewsets, status, mixins
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
from rest_framework.permissions import IsAuthenticated, AllowAny
|
|
from .models import (ComPortReading, User, Report, ScaleProtocol, License,
|
|
StationSlot, generate_doc_number)
|
|
from .serializers import (
|
|
ComPortReadingSerializer, UserSerializer, UserDetailSerializer,
|
|
ChangePasswordSerializer, VehicleSerializer, NomenclatureSerializer,
|
|
NomenclatureEntrySerializer, ReportSerializer, ExtraCombinationSerializer,
|
|
ScaleProtocolSerializer,
|
|
)
|
|
from django.utils import timezone
|
|
from vehicles.models import Vehicle, ExtraCombination, combination_hash
|
|
from nomenclatures.models import Nomenclature, NomenclatureEntry
|
|
from scalesapp.sse import sse_broadcast_update
|
|
from . import licensing
|
|
|
|
|
|
class IsAdminUser(IsAuthenticated):
|
|
def has_permission(self, request, view):
|
|
return bool(request.user and request.user.is_authenticated and request.user.is_admin)
|
|
|
|
|
|
class UserViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
API endpoint for user management.
|
|
|
|
list: Get all users
|
|
create: Create a new user
|
|
retrieve: Get a specific user
|
|
update: Update a user
|
|
destroy: Soft-delete a user (sets is_active=False)
|
|
me: Get current authenticated user
|
|
change_password: Change password for current user
|
|
"""
|
|
queryset = User.objects.all()
|
|
serializer_class = UserSerializer
|
|
filterset_fields = ['role', 'is_admin', 'is_active']
|
|
ordering = ['username']
|
|
|
|
def get_permissions(self):
|
|
if self.action in ('me', 'change_password'):
|
|
return [IsAuthenticated()]
|
|
return [IsAuthenticated(), IsAdminUser()]
|
|
|
|
def destroy(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
instance.is_active = False
|
|
instance.save()
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
@action(detail=False, methods=['get'])
|
|
def me(self, request):
|
|
"""Get current authenticated user details"""
|
|
serializer = UserDetailSerializer(request.user)
|
|
return Response(serializer.data)
|
|
|
|
@action(detail=False, methods=['post'],
|
|
url_path='change-password', url_name='change_password')
|
|
def change_password(self, request):
|
|
"""Change password for current user"""
|
|
serializer = ChangePasswordSerializer(
|
|
data=request.data,
|
|
context={'request': request}
|
|
)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return Response({
|
|
'detail': 'Password updated successfully'
|
|
}, status=status.HTTP_200_OK)
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
class ComPortReadingViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
API endpoint for COM port readings.
|
|
|
|
list: Get all readings with optional filtering
|
|
create: Create a new reading (allows unauthenticated for serial bridge)
|
|
retrieve: Get a specific reading
|
|
destroy: Delete a reading
|
|
latest: Get the latest reading
|
|
by_port: Get readings for a specific port
|
|
"""
|
|
queryset = ComPortReading.objects.all()
|
|
serializer_class = ComPortReadingSerializer
|
|
filterset_fields = ['port']
|
|
ordering = ['-timestamp']
|
|
|
|
def get_permissions(self):
|
|
"""Allow unauthenticated POST for serial bridge"""
|
|
if self.action == 'create':
|
|
return [AllowAny()]
|
|
return [IsAuthenticated()]
|
|
|
|
@action(detail=False, methods=['get'])
|
|
def latest(self, request):
|
|
"""Get the latest reading"""
|
|
reading = ComPortReading.objects.first()
|
|
if reading:
|
|
serializer = self.get_serializer(reading)
|
|
return Response(serializer.data)
|
|
return Response({'detail': 'No readings yet'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
@action(detail=False, methods=['get'])
|
|
def by_port(self, request):
|
|
"""Get readings for a specific port"""
|
|
port = request.query_params.get('port', None)
|
|
if not port:
|
|
return Response(
|
|
{'detail': 'port parameter is required'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
readings = ComPortReading.objects.filter(port=port)
|
|
serializer = self.get_serializer(readings, many=True)
|
|
return Response(serializer.data)
|
|
|
|
def get_client_ip(self, request):
|
|
"""Get client IP address"""
|
|
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
|
|
if x_forwarded_for:
|
|
ip = x_forwarded_for.split(',')[0]
|
|
else:
|
|
ip = request.META.get('REMOTE_ADDR')
|
|
return ip
|
|
|
|
def perform_create(self, serializer):
|
|
"""Save the reading with client IP"""
|
|
serializer.save(source_ip=self.get_client_ip(self.request))
|
|
|
|
|
|
def _maybe_finalize(vehicle, user):
|
|
"""If both tare and gross are set and net hasn't been calculated yet, compute net and assign doc_number."""
|
|
if vehicle.tare is not None and vehicle.gross is not None and vehicle.net is None:
|
|
vehicle.net = vehicle.gross - vehicle.tare
|
|
vehicle.net_date = timezone.now()
|
|
vehicle.net_user = user
|
|
vehicle.doc_number = generate_doc_number(timezone.now().year)
|
|
vehicle.save()
|
|
|
|
|
|
def _assign_weighing(vehicle, value, user, manual=False, note=None):
|
|
"""Record a weighing. Assigns by weight: lighter becomes tare, heavier becomes gross.
|
|
|
|
The first weighing is parked in tare; the second may push it to gross if the new
|
|
reading is lighter, since trucks may arrive either empty (inbound) or loaded (outbound).
|
|
"""
|
|
now = timezone.now()
|
|
|
|
def apply(prefix):
|
|
setattr(vehicle, prefix, value)
|
|
setattr(vehicle, f'{prefix}_date', now)
|
|
setattr(vehicle, f'{prefix}_user', user)
|
|
setattr(vehicle, f'{prefix}_manual', manual)
|
|
setattr(vehicle, f'{prefix}_manual_user', user if manual else None)
|
|
setattr(vehicle, f'{prefix}_manual_date', now if manual else None)
|
|
setattr(vehicle, f'{prefix}_manual_note', note if manual else None)
|
|
|
|
def move(src_prefix, dst_prefix):
|
|
"""Move a recorded weighing to the other slot, carrying all its provenance."""
|
|
for suffix in ('', '_date', '_user', '_manual', '_manual_user', '_manual_date', '_manual_note'):
|
|
setattr(vehicle, f'{dst_prefix}{suffix}', getattr(vehicle, f'{src_prefix}{suffix}'))
|
|
|
|
if vehicle.tare is not None and vehicle.gross is not None:
|
|
raise ValueError('Vehicle already has both weights.')
|
|
|
|
if vehicle.tare is None and vehicle.gross is None:
|
|
apply('tare')
|
|
elif vehicle.gross is None:
|
|
# tare holds the first weighing - keep the lighter one as tare
|
|
if value >= vehicle.tare:
|
|
apply('gross')
|
|
else:
|
|
move('tare', 'gross')
|
|
apply('tare')
|
|
else:
|
|
# gross set but tare empty (only reachable via direct API use)
|
|
if value <= vehicle.gross:
|
|
apply('tare')
|
|
else:
|
|
move('gross', 'tare')
|
|
apply('gross')
|
|
|
|
vehicle.save()
|
|
_maybe_finalize(vehicle, user)
|
|
|
|
|
|
def _record_extra_combination(extra_data):
|
|
"""Record the lookup-only projection of a vehicle's extra data as a reusable combination."""
|
|
if not extra_data:
|
|
return
|
|
|
|
lookup_codes = set(
|
|
Nomenclature.objects.filter(applies_to='vehicle', kind=Nomenclature.LOOKUP)
|
|
.values_list('code', flat=True)
|
|
)
|
|
combo = {k: v for k, v in extra_data.items() if k in lookup_codes and v is not None}
|
|
if not combo:
|
|
return
|
|
|
|
existing, created = ExtraCombination.objects.get_or_create(
|
|
data_hash=combination_hash(combo), defaults={'data': combo}
|
|
)
|
|
if not created:
|
|
existing.use_count += 1
|
|
existing.last_used = timezone.now()
|
|
existing.save()
|
|
|
|
action = 'created' if created else 'updated'
|
|
sse_broadcast_update('extra_combination', action, ExtraCombinationSerializer(existing).data)
|
|
|
|
|
|
class VehicleViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
API endpoint for vehicle management.
|
|
|
|
list: Get all vehicles with pagination
|
|
create: Create a new vehicle with optional extra data
|
|
retrieve: Get a specific vehicle by ID
|
|
update: Full update of vehicle and extra data
|
|
partial_update: Partial update of vehicle
|
|
destroy: Delete a vehicle (cascades to VehicleExtra)
|
|
by_number: Get vehicle by vehicle_number
|
|
measure: Record a weighing (app assigns tare vs gross by weight)
|
|
"""
|
|
queryset = Vehicle.objects.select_related('extra', 'tare_user', 'gross_user').all()
|
|
serializer_class = VehicleSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filterset_fields = ['vehicle_number']
|
|
search_fields = ['vehicle_number']
|
|
ordering = ['vehicle_number']
|
|
|
|
def perform_create(self, serializer):
|
|
instance = serializer.save()
|
|
data = VehicleSerializer(instance).data
|
|
sse_broadcast_update('vehicle', 'created', data)
|
|
_record_extra_combination(instance.extra.data if hasattr(instance, 'extra') else None)
|
|
|
|
def perform_update(self, serializer):
|
|
instance = serializer.save()
|
|
data = VehicleSerializer(instance).data
|
|
sse_broadcast_update('vehicle', 'updated', data)
|
|
_record_extra_combination(instance.extra.data if hasattr(instance, 'extra') else None)
|
|
|
|
def perform_destroy(self, instance):
|
|
data = VehicleSerializer(instance).data
|
|
instance.delete()
|
|
sse_broadcast_update('vehicle', 'deleted', data)
|
|
|
|
@action(detail=False, methods=['get'], url_path='by-number')
|
|
def by_number(self, request):
|
|
"""Get vehicle by vehicle_number query parameter"""
|
|
vehicle_number = request.query_params.get('vehicle_number', None)
|
|
if not vehicle_number:
|
|
return Response(
|
|
{'detail': 'vehicle_number parameter is required'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
try:
|
|
vehicle = Vehicle.objects.get(vehicle_number=vehicle_number)
|
|
serializer = self.get_serializer(vehicle)
|
|
return Response(serializer.data)
|
|
except Vehicle.DoesNotExist:
|
|
return Response(
|
|
{'detail': 'Vehicle not found'},
|
|
status=status.HTTP_404_NOT_FOUND
|
|
)
|
|
|
|
@action(detail=True, methods=['post'], url_path='measure')
|
|
def measure(self, request, pk=None):
|
|
"""Record a weighing. The app assigns tare vs gross by weight."""
|
|
if not request.user.can_measure:
|
|
return Response({'detail': 'You do not have permission to measure.'},
|
|
status=status.HTTP_403_FORBIDDEN)
|
|
|
|
# Licence enforcement. The UI also disables the button, but this is the
|
|
# actual rule - the button state is only a convenience.
|
|
station_key = (request.data.get('station_key') or '').strip()
|
|
if not _station_allowed(station_key):
|
|
limit = licensing.effective_station_limit()
|
|
return Response(
|
|
{'detail': f'This station is not licensed to measure '
|
|
f'(limit: {limit} stations).'},
|
|
status=status.HTTP_403_FORBIDDEN)
|
|
|
|
manual = bool(request.data.get('manual', False))
|
|
if manual and not request.user.can_manually_measure:
|
|
return Response({'detail': 'You do not have permission to measure manually.'},
|
|
status=status.HTTP_403_FORBIDDEN)
|
|
|
|
note = (request.data.get('note') or '').strip()
|
|
if manual and not note:
|
|
return Response({'detail': 'A note is required for a manual measurement.'},
|
|
status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
value = request.data.get('value')
|
|
if value is None:
|
|
return Response({'detail': 'value is required'}, status=status.HTTP_400_BAD_REQUEST)
|
|
try:
|
|
value = int(value)
|
|
except (ValueError, TypeError):
|
|
return Response({'detail': 'value must be a number'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
vehicle = self.get_object()
|
|
try:
|
|
_assign_weighing(vehicle, value, request.user, manual=manual, note=note or None)
|
|
except ValueError as exc:
|
|
return Response({'detail': str(exc)}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
serializer = self.get_serializer(vehicle)
|
|
sse_broadcast_update('vehicle', 'updated', serializer.data)
|
|
return Response(serializer.data)
|
|
|
|
|
|
class ExtraCombinationViewSet(mixins.ListModelMixin,
|
|
mixins.RetrieveModelMixin,
|
|
mixins.DestroyModelMixin,
|
|
viewsets.GenericViewSet):
|
|
"""
|
|
API endpoint for previously-used combinations of vehicle extra (lookup) data.
|
|
|
|
list: Get all combinations, most recently used first
|
|
retrieve: Get a specific combination
|
|
destroy: Delete a combination the user no longer needs
|
|
|
|
Rows are created/bumped automatically when a vehicle's extra data is saved.
|
|
"""
|
|
queryset = ExtraCombination.objects.all()
|
|
serializer_class = ExtraCombinationSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
ordering = ['-last_used']
|
|
|
|
def perform_destroy(self, instance):
|
|
data = ExtraCombinationSerializer(instance).data
|
|
instance.delete()
|
|
sse_broadcast_update('extra_combination', 'deleted', data)
|
|
|
|
|
|
def _live_slots():
|
|
return StationSlot.objects.filter(last_seen__gte=licensing.live_slot_cutoff())
|
|
|
|
|
|
def _station_allowed(station_key):
|
|
"""Whether this station may measure.
|
|
|
|
A station already holding a live slot is always allowed - including when the
|
|
licence is full - so a brief disconnection does not lock a working station
|
|
out behind its own expiring slot.
|
|
"""
|
|
if not station_key:
|
|
return False
|
|
if _live_slots().filter(station_key=station_key).exists():
|
|
return True
|
|
return _live_slots().count() < licensing.effective_station_limit()
|
|
|
|
|
|
class LicenseView(APIView):
|
|
"""Licence status, and installing a new licence key."""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request):
|
|
status_info = licensing.license_status()
|
|
slots = _live_slots()
|
|
payload = {
|
|
**status_info,
|
|
'stations_used': slots.count(),
|
|
'components': licensing.machine_components(),
|
|
}
|
|
if request.user.is_admin:
|
|
payload['slots'] = [
|
|
{'station_key': s.station_key, 'com_port': s.com_port,
|
|
'bridge_version': s.bridge_version, 'last_seen': s.last_seen}
|
|
for s in slots
|
|
]
|
|
return Response(payload)
|
|
|
|
def post(self, request):
|
|
if not request.user.is_admin:
|
|
return Response({'detail': 'Only administrators can install a licence.'},
|
|
status=status.HTTP_403_FORBIDDEN)
|
|
|
|
blob = (request.data.get('key_blob') or '').strip()
|
|
if not blob:
|
|
return Response({'detail': 'key_blob is required'},
|
|
status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Validate before storing, so a bad paste cannot displace a working licence.
|
|
valid, payload, reason = licensing.verify_license(blob)
|
|
if not valid:
|
|
return Response({'detail': reason}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
record = License.objects.first() or License()
|
|
record.key_blob = blob
|
|
record.grace_started = None
|
|
record.save()
|
|
|
|
return Response(licensing.license_status())
|
|
|
|
|
|
class LicenseHeartbeatView(APIView):
|
|
"""Stations report in here so the server can count them.
|
|
|
|
The bridge deliberately does not talk to Django - the browser, which already
|
|
talks to both, reports which bridge it is using. That keeps the bridge pure
|
|
transport with no startup dependency on the server.
|
|
"""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def post(self, request):
|
|
station_key = (request.data.get('station_key') or '').strip()
|
|
if not station_key:
|
|
return Response({'detail': 'station_key is required'},
|
|
status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
limit = licensing.effective_station_limit()
|
|
existing = StationSlot.objects.filter(station_key=station_key).first()
|
|
live = _live_slots()
|
|
|
|
if existing is None and live.count() >= limit:
|
|
return Response({
|
|
'allowed': False,
|
|
'stations_used': live.count(),
|
|
'stations_licensed': limit,
|
|
'detail': f'Licence limit reached ({limit} stations)',
|
|
}, status=status.HTTP_403_FORBIDDEN)
|
|
|
|
slot, _ = StationSlot.objects.update_or_create(
|
|
station_key=station_key,
|
|
defaults={
|
|
'bridge_version': request.data.get('bridge_version', '') or '',
|
|
'com_port': request.data.get('com_port', '') or '',
|
|
},
|
|
)
|
|
# update_or_create does not touch auto_now when nothing changed
|
|
slot.save(update_fields=['last_seen'])
|
|
|
|
return Response({
|
|
'allowed': True,
|
|
'stations_used': _live_slots().count(),
|
|
'stations_licensed': limit,
|
|
})
|
|
|
|
|
|
class ScaleProtocolViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
API endpoint for scale protocol definitions.
|
|
|
|
The frontend caches these and uses them to parse raw serial lines, so adding
|
|
a scale type needs no change to the serial bridge running on station PCs.
|
|
Built-in protocols cannot be deleted.
|
|
"""
|
|
queryset = ScaleProtocol.objects.all()
|
|
serializer_class = ScaleProtocolSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
search_fields = ['name']
|
|
ordering = ['name']
|
|
|
|
def perform_create(self, serializer):
|
|
instance = serializer.save()
|
|
sse_broadcast_update('scale_protocol', 'created', ScaleProtocolSerializer(instance).data)
|
|
|
|
def perform_update(self, serializer):
|
|
instance = serializer.save()
|
|
sse_broadcast_update('scale_protocol', 'updated', ScaleProtocolSerializer(instance).data)
|
|
|
|
def destroy(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
if instance.is_builtin:
|
|
return Response({'detail': 'Built-in protocols cannot be deleted.'},
|
|
status=status.HTTP_400_BAD_REQUEST)
|
|
data = ScaleProtocolSerializer(instance).data
|
|
instance.delete()
|
|
sse_broadcast_update('scale_protocol', 'deleted', data)
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
class NomenclatureViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
API endpoint for nomenclature management.
|
|
|
|
list: Get all nomenclatures with pagination
|
|
create: Create a new nomenclature with fields
|
|
retrieve: Get a specific nomenclature by ID
|
|
update: Full update of nomenclature and fields
|
|
partial_update: Partial update of nomenclature
|
|
destroy: Delete a nomenclature (cascades to fields and entries)
|
|
by_code: Get nomenclature by code
|
|
by_applies_to: Filter nomenclatures by applies_to type
|
|
entries: List or create entries for a specific nomenclature
|
|
"""
|
|
queryset = Nomenclature.objects.prefetch_related('fields').all()
|
|
serializer_class = NomenclatureSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filterset_fields = ['applies_to', 'code', 'kind']
|
|
search_fields = ['code', 'name']
|
|
ordering = ['sort_order', 'code']
|
|
|
|
def perform_create(self, serializer):
|
|
instance = serializer.save()
|
|
sse_broadcast_update('nomenclature', 'insert', NomenclatureSerializer(instance).data)
|
|
|
|
def perform_update(self, serializer):
|
|
instance = serializer.save()
|
|
sse_broadcast_update('nomenclature', 'update', NomenclatureSerializer(instance).data)
|
|
|
|
def perform_destroy(self, instance):
|
|
data = NomenclatureSerializer(instance).data
|
|
instance.delete()
|
|
sse_broadcast_update('nomenclature', 'delete', data)
|
|
|
|
@action(detail=False, methods=['get'], url_path='by-code')
|
|
def by_code(self, request):
|
|
"""Get nomenclature by code query parameter"""
|
|
code = request.query_params.get('code', None)
|
|
if not code:
|
|
return Response(
|
|
{'detail': 'code parameter is required'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
try:
|
|
nomenclature = Nomenclature.objects.get(code=code)
|
|
serializer = self.get_serializer(nomenclature)
|
|
return Response(serializer.data)
|
|
except Nomenclature.DoesNotExist:
|
|
return Response(
|
|
{'detail': 'Nomenclature not found'},
|
|
status=status.HTTP_404_NOT_FOUND
|
|
)
|
|
|
|
@action(detail=False, methods=['get'], url_path='by-applies-to')
|
|
def by_applies_to(self, request):
|
|
"""Get nomenclatures filtered by applies_to query parameter"""
|
|
applies_to = request.query_params.get('applies_to', None)
|
|
if not applies_to:
|
|
return Response(
|
|
{'detail': 'applies_to parameter is required'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
if applies_to not in ['vehicle', 'container']:
|
|
return Response(
|
|
{'detail': 'applies_to must be either vehicle or container'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
nomenclatures = Nomenclature.objects.filter(applies_to=applies_to)
|
|
page = self.paginate_queryset(nomenclatures)
|
|
if page is not None:
|
|
serializer = self.get_serializer(page, many=True)
|
|
return self.get_paginated_response(serializer.data)
|
|
|
|
serializer = self.get_serializer(nomenclatures, many=True)
|
|
return Response(serializer.data)
|
|
|
|
@action(detail=True, methods=['get', 'post'], url_path='entries')
|
|
def entries(self, request, pk=None):
|
|
"""List or create entries for a specific nomenclature."""
|
|
nomenclature = self.get_object()
|
|
|
|
if request.method == 'GET':
|
|
entries = nomenclature.entries.all()
|
|
is_active = request.query_params.get('is_active')
|
|
if is_active is not None:
|
|
entries = entries.filter(is_active=is_active.lower() == 'true')
|
|
serializer = NomenclatureEntrySerializer(entries, many=True)
|
|
return Response(serializer.data)
|
|
|
|
serializer = NomenclatureEntrySerializer(
|
|
data=request.data,
|
|
context={'nomenclature': nomenclature},
|
|
)
|
|
serializer.is_valid(raise_exception=True)
|
|
serializer.save(nomenclature=nomenclature)
|
|
entry_data = {**serializer.data, 'nomenclature_code': nomenclature.code}
|
|
sse_broadcast_update('nomenclature_entry', 'insert', entry_data)
|
|
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
|
|
|
|
|
class NomenclatureEntryViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
API endpoint for individual nomenclature entry management.
|
|
For list/create use the nested endpoint: /api/nomenclatures/{id}/entries/
|
|
This viewset handles retrieve/update/delete of individual entries.
|
|
"""
|
|
queryset = NomenclatureEntry.objects.select_related('nomenclature').all()
|
|
serializer_class = NomenclatureEntrySerializer
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get_serializer_context(self):
|
|
context = super().get_serializer_context()
|
|
if self.kwargs.get('pk'):
|
|
try:
|
|
entry = NomenclatureEntry.objects.select_related('nomenclature').get(pk=self.kwargs['pk'])
|
|
context['nomenclature'] = entry.nomenclature
|
|
except NomenclatureEntry.DoesNotExist:
|
|
pass
|
|
return context
|
|
|
|
def perform_update(self, serializer):
|
|
instance = serializer.save()
|
|
entry_data = {**NomenclatureEntrySerializer(instance).data, 'nomenclature_code': instance.nomenclature.code}
|
|
sse_broadcast_update('nomenclature_entry', 'update', entry_data)
|
|
|
|
def perform_destroy(self, instance):
|
|
entry_data = {**NomenclatureEntrySerializer(instance).data, 'nomenclature_code': instance.nomenclature.code}
|
|
instance.delete()
|
|
sse_broadcast_update('nomenclature_entry', 'delete', entry_data)
|
|
|
|
|
|
class ReportViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
API endpoint for report template management.
|
|
|
|
list: Get all reports
|
|
create: Create a new report
|
|
retrieve: Get a specific report by ID
|
|
update: Update a report
|
|
destroy: Delete a report
|
|
"""
|
|
queryset = Report.objects.select_related('created_by').all()
|
|
serializer_class = ReportSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filterset_fields = ['name']
|
|
search_fields = ['name']
|
|
ordering = ['-updated_at']
|
|
|
|
def perform_create(self, serializer):
|
|
serializer.save(created_by=self.request.user)
|
|
|
|
def perform_update(self, serializer):
|
|
serializer.save()
|