Coverage for backend/django/core/serializer_base.py: 91%

1from rest_framework import serializers 

2from drf_spectacular.types import OpenApiTypes 

3from drf_spectacular.utils import extend_schema_field 

4 

5from core.validation import get_current_flowsheet 

6 

7 

8@extend_schema_field(OpenApiTypes.INT) 

9class StableFlowsheetIdentityField(serializers.Field): 

10 """Accept and return public stable identity without persisting a duplicate FK.""" 

11 

12 def get_attribute(self, instance): 

13 """Pass the full state-owned instance to ``to_representation``.""" 

14 

15 return instance 

16 

17 def to_representation(self, value) -> int: 

18 """Return stable identity without loading the state once per row.""" 

19 

20 context = get_current_flowsheet() or {} 

21 active_state_id = context.get("flowsheet_state") 

22 active_flowsheet_id = context.get("flowsheet") 

23 if ( 

24 active_state_id is not None 

25 and active_flowsheet_id is not None 

26 and value.flowsheet_state_id == int(active_state_id) 

27 ): 

28 return int(active_flowsheet_id) 

29 

30 return value.flowsheet_state.flowsheet_id 

31 

32 def to_internal_value(self, data) -> dict: 

33 """Validate a legacy/public flowsheet payload against request context.""" 

34 

35 try: 

36 flowsheet_id = int(data) 

37 except (TypeError, ValueError): 

38 raise serializers.ValidationError("A valid flowsheet ID is required.") 

39 if flowsheet_id < 1: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 raise serializers.ValidationError("A valid flowsheet ID is required.") 

41 

42 context = get_current_flowsheet() or {} 

43 active_flowsheet_id = context.get("flowsheet") 

44 if ( 

45 active_flowsheet_id is not None 

46 and flowsheet_id != int(active_flowsheet_id) 

47 ): 

48 raise serializers.ValidationError( 

49 "Flowsheet must match the active request flowsheet." 

50 ) 

51 

52 # ``source='*'`` merges this mapping into validated data. Returning an 

53 # empty mapping deliberately accepts the public identity without adding 

54 # a nonexistent stable FK to the model constructor. 

55 return {} 

56 

57 

58class StateOwnedModelSerializer(serializers.ModelSerializer): 

59 """Expose stable flowsheet identity while keeping state IDs internal.""" 

60 

61 flowsheet = StableFlowsheetIdentityField(source="*", required=False) 

62 

63 def get_field_names(self, declared_fields, info): 

64 """Replace the database state field with the stable public identifier.""" 

65 

66 field_names = list(super().get_field_names(declared_fields, info)) 

67 if "flowsheet_state" in field_names: 

68 field_names.remove("flowsheet_state") 

69 if "flowsheet" not in field_names: 

70 field_names.append("flowsheet") 

71 return field_names