Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor procurement serializers and centralize equipment policy

Browse Source
This commit is contained in:
lizerk 2026-08-15 11:34:35 +03:00
parent cca52d70d6
commit a0c87e5b26
17 changed files with 690 additions and 518 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

190
backend/api/v1/procurement/equipment_policy.py Normal file
View File

@ -0,0 +1,190 @@
"""Creation and assignment policy shared by equipment API endpoints."""
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
from api.v1.permissions import has_dept_perm
from employees.constants import DeptPerm
from employees.models import Department, EmployeeDepartment, RoleAssignment
class EquipmentCreationLevel(StrEnum):
FULL = "full"
DEPARTMENT_HEAD = "dept_head"
SCOPED = "scoped"
class EquipmentPolicyValidationError(Exception):
"""Structured validation failure produced by the equipment policy."""
def __init__(self, errors: dict[str, str]):
self.errors = errors
super().__init__(next(iter(errors.values())))
def _scoped_candidate_department_ids(user) -> set[int]:
membership_ids = EmployeeDepartment.objects.filter(
employee_id=user.id,
is_active=True,
).values_list("department_id", flat=True)
assignment_ids = RoleAssignment.objects.filter(
employee_id=user.id,
is_active=True,
).values_list("role__department_id", flat=True)
return set(membership_ids).union(assignment_ids)
def _scoped_departments(user) -> tuple[Department, ...]:
candidate_ids = _scoped_candidate_department_ids(user)
if not candidate_ids:
return ()
departments = Department.objects.filter(id__in=candidate_ids).select_related("head")
return tuple(
department
for department in departments
if has_dept_perm(
user,
department.id,
DeptPerm.MANAGE_EQUIPMENT,
)
)
@dataclass(frozen=True)
class EquipmentCreationPolicy:
user: Any
level: EquipmentCreationLevel | None
departments: tuple[Department, ...]
@classmethod
def resolve(cls, user) -> "EquipmentCreationPolicy":
if not user or not user.is_authenticated:
return cls(user=user, level=None, departments=())
if (
user.is_staff
or user.is_superuser
or user.has_perm("procurement.add_equipment")
):
return cls(
user=user,
level=EquipmentCreationLevel.FULL,
departments=(),
)
headed_departments = tuple(
Department.objects.filter(head_id=user.id).select_related("head")
)
if headed_departments:
return cls(
user=user,
level=EquipmentCreationLevel.DEPARTMENT_HEAD,
departments=headed_departments,
)
scoped_departments = _scoped_departments(user)
if scoped_departments:
return cls(
user=user,
level=EquipmentCreationLevel.SCOPED,
departments=scoped_departments,
)
return cls(user=user, level=None, departments=())
def validate_create(self, attributes: dict[str, Any]) -> dict[str, Any]:
values = dict(attributes)
department = values.get("department")
if self.level is EquipmentCreationLevel.FULL:
pass
elif self.level is EquipmentCreationLevel.DEPARTMENT_HEAD:
department = self._use_allowed_department(department)
values["department"] = department
values.setdefault("responsible_person", self.user)
elif self.level is EquipmentCreationLevel.SCOPED:
department = self._use_allowed_department(department)
values["department"] = department
values["responsible_person"] = department.head
else:
raise EquipmentPolicyValidationError(
{"non_field_errors": "У вас нет прав на создание оборудования."}
)
validate_responsible_department(
department=department,
responsible=values.get("responsible_person"),
)
return values
def _use_allowed_department(self, department) -> Department:
allowed_by_id = {item.id: item for item in self.departments}
if department is not None and department.id not in allowed_by_id:
raise EquipmentPolicyValidationError(
{
"department": (
"Вы можете создавать оборудование только в своём отделе."
)
}
)
if department is not None:
return allowed_by_id[department.id]
return self.departments[0]
def as_options(self) -> dict[str, Any]:
if self.level is EquipmentCreationLevel.FULL:
departments = Department.objects.all()
else:
departments = self.departments
default_responsible = None
if self.level is EquipmentCreationLevel.DEPARTMENT_HEAD:
default_responsible = self.user
elif self.level is EquipmentCreationLevel.SCOPED:
default_responsible = next(
(department.head for department in departments if department.head),
None,
)
return {
"allowed_departments": [
{"id": department.id, "name": department.name}
for department in departments
],
"can_choose_department": self.level is EquipmentCreationLevel.FULL,
"can_choose_responsible": self.level
in {
EquipmentCreationLevel.FULL,
EquipmentCreationLevel.DEPARTMENT_HEAD,
},
"default_responsible": (
{
"id": default_responsible.id,
"name": default_responsible.get_full_name(),
}
if default_responsible
else None
),
"permission_level": self.level.value if self.level else None,
}
def validate_responsible_department(*, department, responsible) -> None:
if not department or not responsible:
return
if department.head_id == responsible.id:
return
if EmployeeDepartment.objects.filter(
employee_id=responsible.id,
department_id=department.id,
is_active=True,
).exists():
return
raise EquipmentPolicyValidationError(
{
"responsible_person": (
"Ответственный должен состоять в том же отделе, что и оборудование."
)
}
)

2
backend/api/v1/procurement/serializers/__init__.py
View File

@ -21,6 +21,7 @@ from .requests import (
ProcurementRequestCreateSerializer,
ProcurementRequestDetailSerializer,
ProcurementRequestListSerializer,
ProcurementRequestPermissionFieldsMixin,
ProcurementRequestSummaryMixin,
)
from .responses import (
@ -51,6 +52,7 @@ __all__ = [
"ProcurementRequestCreateSerializer",
"ProcurementRequestDetailSerializer",
"ProcurementRequestListSerializer",
"ProcurementRequestPermissionFieldsMixin",
"ProcurementRequestSummaryMixin",
"SupplierSerializer",
]

1
backend/api/v1/procurement/serializers/activity.py
View File

@ -5,6 +5,7 @@ from rest_framework import serializers
from procurement.models import ProcurementRequestActivity
from ...employees.serializers import EmployeeBriefSerializer
class ProcurementRequestActivitySerializer(serializers.ModelSerializer):
actor = EmployeeBriefSerializer(read_only=True)
action_display = serializers.CharField(

13
backend/api/v1/procurement/serializers/approvals.py
View File

@ -10,6 +10,7 @@ from ...employees.serializers import DepartmentBriefSerializer, EmployeeBriefSer
User = get_user_model()
class ProcurementRequestCreateOptionsSerializer(serializers.Serializer):
"""Опции формы создания заявки на закупку."""
@ -32,9 +33,7 @@ class ProcurementManualApprovalStepSerializer(serializers.Serializer):
def validate_priority(self, value):
if not ApprovalRoute.objects.filter(priority=value).exists():
raise serializers.ValidationError(
"Выберите настроенный этап согласования."
)
raise serializers.ValidationError("Выберите настроенный этап согласования.")
return value
@ -50,9 +49,7 @@ class ProcurementManualApprovalStepsSerializer(serializers.Serializer):
def validate_approval_steps(self, value):
priorities = [item["priority"] for item in value]
duplicate_priorities = {
priority
for priority in priorities
if priorities.count(priority) > 1
priority for priority in priorities if priorities.count(priority) > 1
}
if duplicate_priorities:
raise serializers.ValidationError(
@ -68,9 +65,7 @@ class ApprovalSerializer(serializers.ModelSerializer):
approver_name = serializers.CharField(
source="approver.get_full_name", read_only=True
)
status_display = serializers.CharField(
source="get_status_display", read_only=True
)
status_display = serializers.CharField(source="get_status_display", read_only=True)
step_label = serializers.SerializerMethodField()
@extend_schema_field(serializers.CharField())

5
backend/api/v1/procurement/serializers/budgets.py
View File

@ -4,12 +4,11 @@ from rest_framework import serializers
from procurement.models import Budget
class BudgetSerializer(serializers.ModelSerializer):
"""Сериализатор бюджета отдела."""
department_name = serializers.CharField(
source="department.name", read_only=True
)
department_name = serializers.CharField(source="department.name", read_only=True)
remaining_amount = serializers.DecimalField(
max_digits=14, decimal_places=2, read_only=True
)

1
backend/api/v1/procurement/serializers/categories.py
View File

@ -5,6 +5,7 @@ from rest_framework import serializers
from procurement.models import EquipmentCategory
class EquipmentCategorySerializer(serializers.ModelSerializer):
"""Сериализатор для категорий оборудования."""

265
backend/api/v1/procurement/serializers/equipment.py
View File

@ -4,25 +4,27 @@ from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
from procurement.models import Equipment
from procurement.services import InventoryNumberGenerator
from ..equipment_policy import (
EquipmentCreationPolicy,
EquipmentPolicyValidationError,
validate_responsible_department,
)
class EquipmentListSerializer(serializers.ModelSerializer):
"""Сериализатор для списка оборудования."""
category_name = serializers.CharField(
source="category.name", read_only=True
)
category_name = serializers.CharField(source="category.name", read_only=True)
category_icon = serializers.CharField(
source="category.icon", read_only=True, default="bi-box-seam"
)
department_name = serializers.CharField(
source="department.name", read_only=True
)
department_name = serializers.CharField(source="department.name", read_only=True)
responsible_name = serializers.CharField(
source="responsible_person.get_full_name", read_only=True
)
status_display = serializers.CharField(
source="get_status_display", read_only=True
)
status_display = serializers.CharField(source="get_status_display", read_only=True)
is_under_warranty = serializers.BooleanField(read_only=True)
comments_count = serializers.IntegerField(read_only=True, default=0)
@ -67,250 +69,41 @@ class EquipmentListSerializer(serializers.ModelSerializer):
def create(self, validated_data):
"""Автоматическая генерация инвентарного номера при создании."""
import re
from datetime import datetime
year = datetime.now().year
prefix = f"INV-{year}-"
# Находим максимальный номер с текущим префиксом
last_equipment = (
Equipment.objects.filter(inventory_number__startswith=prefix)
.order_by("-inventory_number")
.first()
)
if last_equipment:
match = re.search(r"(\d+)$", last_equipment.inventory_number)
if match:
next_num = int(match.group(1)) + 1
else:
next_num = 1
else:
next_num = 1
validated_data["inventory_number"] = f"{prefix}{next_num:04d}"
validated_data["inventory_number"] = InventoryNumberGenerator.generate()
return super().create(validated_data)
def _get_user_permission_level(self, user):
"""Определяет уровень прав пользователя.
Возвращает:
'full' - админ/модельные права
(свободный выбор отдела/ответственного)
'dept_head' - начальник отдела
(свой отдел, выбор ответственного)
'scoped' - скоуп-право (свой отдел, ответственный = начальник)
None - нет прав на создание
"""
from api.v1.permissions import has_dept_perm
from employees.constants import DeptPerm
from employees.models import Department, EmployeeDepartment
if not user or not user.is_authenticated:
return None
# Админ или модельные права = полный доступ
if (
user.is_staff
or user.is_superuser
or user.has_perm("procurement.add_equipment")
):
return "full"
# Проверяем, является ли начальником какого-то отдела
headed_depts = Department.objects.filter(head_id=user.id)
if headed_depts.exists():
return "dept_head"
# Проверяем скоуп-право в активных отделах
user_dept_links = EmployeeDepartment.objects.filter(
employee_id=user.id, is_active=True
).select_related("department", "role")
for link in user_dept_links:
if has_dept_perm(
user, link.department_id, DeptPerm.MANAGE_EQUIPMENT
):
return "scoped"
return None
def _get_user_allowed_departments(self, user, perm_level):
"""Возвращает отделы, где пользователь может создать оборудование."""
from employees.models import Department, EmployeeDepartment
if perm_level == "full":
return list(Department.objects.all())
if perm_level == "dept_head":
return list(Department.objects.filter(head_id=user.id))
if perm_level == "scoped":
from api.v1.permissions import has_dept_perm
from employees.constants import DeptPerm
result = []
user_dept_links = EmployeeDepartment.objects.filter(
employee_id=user.id, is_active=True
).select_related("department")
for link in user_dept_links:
if has_dept_perm(
user, link.department_id, DeptPerm.MANAGE_EQUIPMENT
):
result.append(link.department)
return result
return []
def validate(self, attrs):
"""Проверка и автозаполнение полей в зависимости от прав.
Уровни прав:
- full (админ/модельные права): свободный выбор отдела
и ответственного
- dept_head (начальник): только свой отдел,
выбор ответственного из отдела
- scoped (скоуп-право): только свой отдел, ответственный = начальник
"""
"""Validate department access and responsible-person consistency."""
request = self.context.get("request")
user = request.user if request else None
# При update (есть instance) используем облегчённую валидацию
if self.instance is not None:
return self._validate_update(attrs, user)
try:
if self.instance is None:
return EquipmentCreationPolicy.resolve(user).validate_create(attrs)
# Создание — полная проверка прав
perm_level = self._get_user_permission_level(user)
# Проверяем отдел
department = attrs.get("department")
if perm_level == "full":
# Полный доступ — можно указать любой отдел
pass
elif perm_level == "dept_head":
# Начальник — только свои отделы
allowed_depts = self._get_user_allowed_departments(user, perm_level)
allowed_ids = [d.id for d in allowed_depts]
if department and department.id not in allowed_ids:
raise serializers.ValidationError(
{
"department": "Вы можете создавать оборудование "
"только в своём отделе."
}
)
# Если отдел не указан — берём первый
if not department and allowed_depts:
attrs["department"] = allowed_depts[0]
department = allowed_depts[0]
# Ответственный по умолчанию — сам начальник
resp = attrs.get("responsible_person")
if resp is None:
attrs["responsible_person"] = user
elif perm_level == "scoped":
# Скоуп-право — только свой отдел, ответственный = начальник
allowed_depts = self._get_user_allowed_departments(user, perm_level)
allowed_ids = [d.id for d in allowed_depts]
if department and department.id not in allowed_ids:
raise serializers.ValidationError(
{
"department": "Вы можете создавать оборудование "
"только в своём отделе."
}
)
# Если отдел не указан — берём первый
if not department and allowed_depts:
attrs["department"] = allowed_depts[0]
department = allowed_depts[0]
# Ответственный ВСЕГДА = начальник отдела (без выбора)
if department and department.head:
attrs["responsible_person"] = department.head
else:
attrs["responsible_person"] = None
else:
# Нет прав — ошибка
raise serializers.ValidationError(
{"non_field_errors": "У вас нет прав на создание оборудования."}
department = attrs.get("department", self.instance.department)
responsible = attrs.get(
"responsible_person",
self.instance.responsible_person,
)
# Валидация ответственного
return self._validate_responsible_in_department(attrs)
def _validate_update(self, attrs, user):
"""Валидация при обновлении (облегчённая)."""
# При update не проверяем права создания,
# только валидацию ответственного если он меняется
department = attrs.get("department", self.instance.department)
responsible = attrs.get("responsible_person")
# Если меняется ответственный — проверяем принадлежность к отделу
if responsible is not None:
attrs["department"] = department # Для валидации
attrs["responsible_person"] = responsible
return self._validate_responsible_in_department(attrs)
return attrs
def _validate_responsible_in_department(self, attrs):
"""Проверяет, что ответственный состоит в отделе оборудования."""
from employees.models import EmployeeDepartment
department = attrs.get("department")
responsible = attrs.get("responsible_person")
if department and responsible:
# Проверяем, что ответственный в этом отделе
is_in_dept = (
# Он начальник отдела
department.head_id == responsible.id
or
# Или состоит в отделе
EmployeeDepartment.objects.filter(
employee_id=responsible.id,
department_id=department.id,
is_active=True,
).exists()
validate_responsible_department(
department=department,
responsible=responsible,
)
if not is_in_dept:
raise serializers.ValidationError(
{
"responsible_person": (
"Ответственный должен состоять в том же отделе, "
"что и оборудование."
)
}
)
return attrs
return attrs
except EquipmentPolicyValidationError as exc:
raise serializers.ValidationError(exc.errors) from exc
class EquipmentDetailSerializer(serializers.ModelSerializer):
"""Сериализатор для детальной информации об оборудовании."""
category_name = serializers.CharField(
source="category.name", read_only=True
)
department_name = serializers.CharField(
source="department.name", read_only=True
)
category_name = serializers.CharField(source="category.name", read_only=True)
department_name = serializers.CharField(source="department.name", read_only=True)
responsible_name = serializers.CharField(
source="responsible_person.get_full_name", read_only=True
)
status_display = serializers.CharField(
source="get_status_display", read_only=True
)
status_display = serializers.CharField(source="get_status_display", read_only=True)
is_under_warranty = serializers.BooleanField(read_only=True)
maintenance_count = serializers.SerializerMethodField()

9
backend/api/v1/procurement/serializers/items.py
View File

@ -8,6 +8,7 @@ from procurement.models import ProcurementItem, ProcurementItemAttachment
from .common import _validate_expected_delivery_dates
class ProcurementItemAttachmentSerializer(serializers.ModelSerializer):
uploaded_by_name = serializers.CharField(
source="uploaded_by.get_full_name",
@ -32,10 +33,7 @@ class ProcurementItemAttachmentSerializer(serializers.ModelSerializer):
read_only_fields = fields
def get_download_url(self, obj) -> str:
return (
f"/api/v1/procurement/items/{obj.item_id}/"
f"attachments/{obj.id}/"
)
return f"/api/v1/procurement/items/{obj.item_id}/attachments/{obj.id}/"
def get_thumbnail_url(self, obj) -> str | None:
from procurement.media import procurement_attachment_media_kind
@ -43,8 +41,7 @@ class ProcurementItemAttachmentSerializer(serializers.ModelSerializer):
if not procurement_attachment_media_kind(obj):
return None
return (
f"/api/v1/procurement/items/{obj.item_id}/"
f"attachments/{obj.id}/thumbnail/"
f"/api/v1/procurement/items/{obj.item_id}/attachments/{obj.id}/thumbnail/"
)

9
backend/api/v1/procurement/serializers/maintenance.py
View File

@ -4,21 +4,18 @@ from rest_framework import serializers
from procurement.models import MaintenanceRecord
class MaintenanceRecordSerializer(serializers.ModelSerializer):
"""Сериализатор для записей обслуживания."""
equipment_name = serializers.CharField(
source="equipment.name", read_only=True
)
equipment_name = serializers.CharField(source="equipment.name", read_only=True)
equipment_inventory = serializers.CharField(
source="equipment.inventory_number", read_only=True
)
performed_by_name = serializers.CharField(
source="performed_by.get_full_name", read_only=True
)
type_display = serializers.CharField(
source="get_type_display", read_only=True
)
type_display = serializers.CharField(source="get_type_display", read_only=True)
class Meta:
model = MaintenanceRecord

238
backend/api/v1/procurement/serializers/requests.py
View File

@ -1,12 +1,15 @@
"""Procurement request serializers."""
from django.db import transaction
from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
from communications import comments_helpers
from procurement.constants import ProcurementItemExecutionStatus, ProcurementStatus
from procurement.models import ProcurementItem, ProcurementRequest, ProcurementRequestView
from procurement.application import (
ProcurementRequestItemNotFound,
create_procurement_request,
update_procurement_request_with_items,
)
from procurement.constants import ProcurementItemExecutionStatus
from procurement.models import ProcurementRequest, ProcurementRequestView
from procurement.services import ProcurementApprovalResolver
from .approvals import ApprovalSerializer
@ -26,6 +29,7 @@ from .items import (
ProcurementItemSerializer,
)
class ProcurementRequestSummaryMixin(serializers.Serializer):
is_viewed = serializers.SerializerMethodField()
last_arrival_notice_at = serializers.DateTimeField(
@ -97,21 +101,24 @@ class ProcurementRequestSummaryMixin(serializers.Serializer):
@extend_schema_field(serializers.IntegerField())
def get_items_received_count(self, obj):
return sum(
1 for item in _request_items(obj)
1
for item in _request_items(obj)
if item.execution_status == ProcurementItemExecutionStatus.RECEIVED
)
@extend_schema_field(serializers.IntegerField())
def get_items_problem_count(self, obj):
return sum(
1 for item in _request_items(obj)
1
for item in _request_items(obj)
if item.execution_status in PROBLEM_ITEM_STATUSES
)
@extend_schema_field(serializers.IntegerField())
def get_items_pending_count(self, obj):
return sum(
1 for item in _request_items(obj)
1
for item in _request_items(obj)
if item.execution_status == ProcurementItemExecutionStatus.PENDING
)
@ -193,8 +200,50 @@ class ProcurementRequestSummaryMixin(serializers.Serializer):
]
class ProcurementRequestPermissionFieldsMixin(serializers.Serializer):
"""Expose request workflow capabilities for the current API user."""
can_current_user_approve = serializers.SerializerMethodField()
can_current_user_submit_for_approval = serializers.SerializerMethodField()
can_current_user_start_work = serializers.SerializerMethodField()
can_current_user_process_items = serializers.SerializerMethodField()
def _current_user(self):
request = self.context.get("request")
return getattr(request, "user", None)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_approve(self, obj):
return ProcurementApprovalResolver.user_can_approve(
self._current_user(),
obj,
)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_submit_for_approval(self, obj):
return ProcurementApprovalResolver.user_can_submit_for_approval(
self._current_user(),
obj,
)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_start_work(self, obj):
return ProcurementApprovalResolver.user_can_start_work(
self._current_user(),
obj,
)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_process_items(self, obj):
return ProcurementApprovalResolver.user_can_process_items(
self._current_user(),
obj,
)
class ProcurementRequestListSerializer(
ProcurementRequestSummaryMixin,
ProcurementRequestPermissionFieldsMixin,
serializers.ModelSerializer,
):
"""Сериализатор для списка заявок (краткий)."""
@ -205,18 +254,14 @@ class ProcurementRequestListSerializer(
required=False,
write_only=True,
)
department_name = serializers.CharField(
source="department.name", read_only=True
)
department_name = serializers.CharField(source="department.name", read_only=True)
processing_department_name = serializers.CharField(
source="processing_department.name", read_only=True, allow_null=True
)
requestor_name = serializers.CharField(
source="requestor.get_full_name", read_only=True
)
status_display = serializers.CharField(
source="get_status_display", read_only=True
)
status_display = serializers.CharField(source="get_status_display", read_only=True)
urgency_display = serializers.CharField(
source="get_urgency_display", read_only=True
)
@ -231,37 +276,6 @@ class ProcurementRequestListSerializer(
source="executor.get_full_name", read_only=True, allow_null=True
)
comments_count = serializers.IntegerField(read_only=True, default=0)
can_current_user_approve = serializers.SerializerMethodField()
can_current_user_submit_for_approval = serializers.SerializerMethodField()
can_current_user_start_work = serializers.SerializerMethodField()
can_current_user_process_items = serializers.SerializerMethodField()
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_approve(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_approve(user, obj)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_submit_for_approval(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_submit_for_approval(
user,
obj,
)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_start_work(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_start_work(user, obj)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_process_items(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_process_items(user, obj)
class Meta:
model = ProcurementRequest
@ -364,43 +378,17 @@ class ProcurementRequestListSerializer(
request = self.context.get("request")
author = getattr(request, "user", None)
with transaction.atomic():
instance = super().update(instance, validated_data)
existing_items = {item.id: item for item in instance.items.all()}
kept_item_ids = set()
for item_data in items_data:
item_id = item_data.pop("id", None)
initial_comment = item_data.pop("initial_comment", "").strip()
if item_id is not None:
item = existing_items.get(item_id)
if item is None:
raise serializers.ValidationError(
{"items": f"Позиция {item_id} не найдена в заявке."}
)
for field, value in item_data.items():
setattr(item, field, value)
item.save()
else:
item = ProcurementItem.objects.create(
request=instance,
**item_data,
)
kept_item_ids.add(item.id)
if initial_comment and author and author.is_authenticated:
comments_helpers.create_comment(
obj=item,
author=author,
content=initial_comment,
)
for item_id, item in existing_items.items():
if item_id not in kept_item_ids:
item.delete()
instance.recalculate_fulfillment_status(save=True)
return instance
try:
return update_procurement_request_with_items(
request_id=instance.id,
attributes=validated_data,
items=items_data,
author=author,
)
except ProcurementRequestItemNotFound as exc:
raise serializers.ValidationError(
{"items": f"Позиция {exc.args[0]} не найдена в заявке."}
) from exc
def validate(self, attrs):
attrs = super().validate(attrs)
@ -421,49 +409,37 @@ class ProcurementRequestListSerializer(
{"items": "Добавьте хотя бы одну позицию."}
)
if self.instance is not None:
valid_item_ids = set(
self.instance.items.values_list("id", flat=True)
)
valid_item_ids = set(self.instance.items.values_list("id", flat=True))
incoming_item_ids = {
item["id"] for item in items_data if item.get("id") is not None
}
unknown_item_ids = incoming_item_ids - valid_item_ids
if unknown_item_ids:
raise serializers.ValidationError(
{
"items": (
"Позиция "
f"{min(unknown_item_ids)} не найдена в заявке."
)
}
{"items": (f"Позиция {min(unknown_item_ids)} не найдена в заявке.")}
)
return attrs
class ProcurementRequestDetailSerializer(
ProcurementRequestSummaryMixin,
ProcurementRequestPermissionFieldsMixin,
serializers.ModelSerializer,
):
"""Сериализатор для детальной информации о заявке."""
department_name = serializers.CharField(
source="department.name", read_only=True
)
department_name = serializers.CharField(source="department.name", read_only=True)
processing_department_name = serializers.CharField(
source="processing_department.name", read_only=True, allow_null=True
)
requestor_name = serializers.CharField(
source="requestor.get_full_name", read_only=True
)
requestor_email = serializers.CharField(
source="requestor.email", read_only=True
)
requestor_email = serializers.CharField(source="requestor.email", read_only=True)
executor_name = serializers.CharField(
source="executor.get_full_name", read_only=True, allow_null=True
)
status_display = serializers.CharField(
source="get_status_display", read_only=True
)
status_display = serializers.CharField(source="get_status_display", read_only=True)
urgency_display = serializers.CharField(
source="get_urgency_display", read_only=True
)
@ -481,37 +457,6 @@ class ProcurementRequestDetailSerializer(
is_editable = serializers.BooleanField(read_only=True)
comments_count = serializers.IntegerField(read_only=True, default=0)
activity_count = serializers.SerializerMethodField()
can_current_user_approve = serializers.SerializerMethodField()
can_current_user_submit_for_approval = serializers.SerializerMethodField()
can_current_user_start_work = serializers.SerializerMethodField()
can_current_user_process_items = serializers.SerializerMethodField()
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_approve(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_approve(user, obj)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_submit_for_approval(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_submit_for_approval(
user,
obj,
)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_start_work(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_start_work(user, obj)
@extend_schema_field(serializers.BooleanField())
def get_can_current_user_process_items(self, obj):
request = self.context.get("request")
user = getattr(request, "user", None)
return ProcurementApprovalResolver.user_can_process_items(user, obj)
@extend_schema_field(serializers.IntegerField())
def get_activity_count(self, obj):
@ -630,9 +575,7 @@ class ProcurementRequestCreateSerializer(serializers.ModelSerializer):
read_only_fields = ["id", "requestor", "status", "created_at"]
def validate(self, attrs):
_validate_processing_department_available(
attrs.get("processing_department")
)
_validate_processing_department_available(attrs.get("processing_department"))
if (
not attrs.get("processing_department")
and not attrs.get("description", "").strip()
@ -646,31 +589,8 @@ class ProcurementRequestCreateSerializer(serializers.ModelSerializer):
"""Создать заявку с позициями."""
items_data = validated_data.pop("items", [])
request = self.context["request"]
has_processing_department = bool(
validated_data.get("processing_department")
)
if has_processing_department:
validated_data["status"] = ProcurementStatus.WAITING
procurement_request = ProcurementRequest(
return create_procurement_request(
requestor=request.user,
**validated_data,
attributes=validated_data,
items=items_data,
)
procurement_request._notification_actor = request.user
procurement_request.save()
# Создаем позиции (если они переданы)
for item_data in items_data:
initial_comment = item_data.pop("initial_comment", "").strip()
item = ProcurementItem.objects.create(
request=procurement_request, **item_data
)
if initial_comment:
comments_helpers.create_comment(
obj=item,
author=request.user,
content=initial_comment,
)
return procurement_request

1
backend/api/v1/procurement/serializers/responses.py
View File

@ -2,6 +2,7 @@
from rest_framework import serializers
class ProcurementNotificationReadResponseSerializer(serializers.Serializer):
status = serializers.CharField()
count = serializers.IntegerField()

1
backend/api/v1/procurement/serializers/suppliers.py
View File

@ -4,6 +4,7 @@ from rest_framework import serializers
from procurement.models import Supplier
class SupplierSerializer(serializers.ModelSerializer):
"""Сериализатор для поставщиков."""

103
backend/api/v1/procurement/viewsets/equipment.py
View File

@ -10,7 +10,7 @@ from rest_framework import filters, status, viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from employees.models import Department, EmployeeDepartment
from employees.models import Department
from procurement.constants import (
EquipmentStatus,
)
@ -26,6 +26,7 @@ from ..permissions import (
CanManageEquipment,
CanManageEquipmentCategory,
)
from ..equipment_policy import EquipmentCreationPolicy
from ..serializers import (
EquipmentCategorySerializer,
EquipmentDetailSerializer,
@ -207,104 +208,8 @@ class EquipmentViewSet(viewsets.ModelViewSet):
- can_choose_responsible: может ли выбирать ответственного
- default_responsible: ответственный по умолчанию (если нет выбора)
"""
from api.v1.permissions import has_dept_perm
from employees.constants import DeptPerm
user = request.user
# Определяем уровень прав
if user.is_staff or user.is_superuser:
perm_level = "full"
elif user.has_perm("procurement.add_equipment"):
perm_level = "full"
elif Department.objects.filter(head_id=user.id).exists():
perm_level = "dept_head"
else:
# Проверяем скоуп-право
user_dept_links = EmployeeDepartment.objects.filter(
employee_id=user.id, is_active=True
).select_related("department")
has_scoped = False
for link in user_dept_links:
if has_dept_perm(user, link.department_id, DeptPerm.MANAGE_EQUIPMENT):
has_scoped = True
break
perm_level = "scoped" if has_scoped else None
# Формируем ответ в зависимости от уровня прав
if perm_level == "full":
# Полный доступ — все отделы
departments = Department.objects.all().values("id", "name")
return Response(
{
"allowed_departments": list(departments),
"can_choose_department": True,
"can_choose_responsible": True,
"default_responsible": None,
"permission_level": "full",
}
)
elif perm_level == "dept_head":
# Начальник — только свои отделы
departments = Department.objects.filter(head_id=user.id).values(
"id", "name"
)
return Response(
{
"allowed_departments": list(departments),
"can_choose_department": False,
"can_choose_responsible": True,
"default_responsible": {
"id": user.id,
"name": user.get_full_name(),
},
"permission_level": "dept_head",
}
)
elif perm_level == "scoped":
# Скоуп-право — отделы с правом, ответственный = начальник
allowed_depts = []
default_responsible = None
user_dept_links = EmployeeDepartment.objects.filter(
employee_id=user.id, is_active=True
).select_related("department", "department__head")
for link in user_dept_links:
if has_dept_perm(user, link.department_id, DeptPerm.MANAGE_EQUIPMENT):
dept = link.department
allowed_depts.append({"id": dept.id, "name": dept.name})
if default_responsible is None and dept.head:
default_responsible = {
"id": dept.head.id,
"name": dept.head.get_full_name(),
}
return Response(
{
"allowed_departments": allowed_depts,
"can_choose_department": False,
"can_choose_responsible": False,
"default_responsible": default_responsible,
"permission_level": "scoped",
}
)
else:
# Нет прав
return Response(
{
"allowed_departments": [],
"can_choose_department": False,
"can_choose_responsible": False,
"default_responsible": None,
"permission_level": None,
}
)
policy = EquipmentCreationPolicy.resolve(request.user)
return Response(policy.as_options())
@action(detail=True, methods=["post"])
def transfer(self, request, pk=None):

10
backend/procurement/application/__init__.py
View File

@ -3,9 +3,19 @@ from .equipment import (
ProcurementItemUnavailable,
create_equipment_from_procurement_item,
)
from .requests import (
ProcurementRequestItemNotFound,
ProcurementRequestUpdateError,
create_procurement_request,
update_procurement_request_with_items,
)
__all__ = [
"EquipmentCreationError",
"ProcurementItemUnavailable",
"ProcurementRequestItemNotFound",
"ProcurementRequestUpdateError",
"create_equipment_from_procurement_item",
"create_procurement_request",
"update_procurement_request_with_items",
]

119
backend/procurement/application/requests.py Normal file
View File

@ -0,0 +1,119 @@
"""Application use cases for creating and editing procurement requests."""
from collections.abc import Mapping, Sequence
from typing import Any
from django.db import transaction
from communications import comments_helpers
from procurement.constants import ProcurementStatus
from procurement.models import ProcurementItem, ProcurementRequest
class ProcurementRequestUpdateError(Exception):
"""Expected conflict while applying a procurement request update."""
class ProcurementRequestItemNotFound(ProcurementRequestUpdateError):
"""An update referenced an item that does not belong to the request."""
def _create_initial_comment(*, item, author, content: str) -> None:
if content and author and author.is_authenticated:
comments_helpers.create_comment(
obj=item,
author=author,
content=content,
)
@transaction.atomic
def create_procurement_request(
*,
requestor,
attributes: Mapping[str, Any],
items: Sequence[Mapping[str, Any]],
) -> ProcurementRequest:
"""Create a request and all initial items as one atomic operation."""
request_attributes = dict(attributes)
if request_attributes.get("processing_department"):
request_attributes["status"] = ProcurementStatus.WAITING
procurement_request = ProcurementRequest(
requestor=requestor,
**request_attributes,
)
procurement_request._notification_actor = requestor
procurement_request.save()
for raw_item in items:
item_attributes = dict(raw_item)
initial_comment = item_attributes.pop("initial_comment", "").strip()
item = ProcurementItem.objects.create(
request=procurement_request,
**item_attributes,
)
_create_initial_comment(
item=item,
author=requestor,
content=initial_comment,
)
return procurement_request
@transaction.atomic
def update_procurement_request_with_items(
*,
request_id: int,
attributes: Mapping[str, Any],
items: Sequence[Mapping[str, Any]],
author,
) -> ProcurementRequest:
"""Update request fields and synchronize its item collection atomically."""
procurement_request = ProcurementRequest.objects.select_for_update().get(
pk=request_id,
)
existing_items = {
item.id: item for item in procurement_request.items.select_for_update().all()
}
for field, value in attributes.items():
setattr(procurement_request, field, value)
procurement_request.save()
kept_item_ids: set[int] = set()
for raw_item in items:
item_attributes = dict(raw_item)
item_id = item_attributes.pop("id", None)
initial_comment = item_attributes.pop("initial_comment", "").strip()
if item_id is None:
item = ProcurementItem.objects.create(
request=procurement_request,
**item_attributes,
)
else:
item = existing_items.get(item_id)
if item is None:
raise ProcurementRequestItemNotFound(item_id)
for field, value in item_attributes.items():
setattr(item, field, value)
item.save()
kept_item_ids.add(item.id)
_create_initial_comment(
item=item,
author=author,
content=initial_comment,
)
obsolete_item_ids = set(existing_items) - kept_item_ids
if obsolete_item_ids:
ProcurementItem.objects.filter(
request=procurement_request,
id__in=obsolete_item_ids,
).delete()
procurement_request.recalculate_fulfillment_status(save=True)
return procurement_request

85
backend/tests/api/v1/procurement/test_equipment_policy.py Normal file
View File

@ -0,0 +1,85 @@
import pytest
from api.v1.procurement.equipment_policy import (
EquipmentCreationLevel,
EquipmentCreationPolicy,
EquipmentPolicyValidationError,
validate_responsible_department,
)
from employees.constants import DeptPerm
from employees.models import (
DepartmentPermission,
DepartmentRole,
RoleAssignment,
)
pytestmark = pytest.mark.django_db
def _grant_equipment_role(*, user, department):
permission, _ = DepartmentPermission.objects.get_or_create(
code=DeptPerm.MANAGE_EQUIPMENT,
defaults={"name": "Управление оборудованием"},
)
role = DepartmentRole.objects.create(
department=department,
name="Управляющий оборудованием",
)
role.scoped_permissions.add(permission)
RoleAssignment.objects.create(employee=user, role=role)
def test_policy_recognizes_role_assignment_without_department_membership(
user_factory,
department_factory,
):
user = user_factory("equipment-role@example.com")
department = department_factory("Инфраструктура")
_grant_equipment_role(user=user, department=department)
policy = EquipmentCreationPolicy.resolve(user)
assert policy.level is EquipmentCreationLevel.SCOPED
assert policy.departments == (department,)
assert policy.as_options()["allowed_departments"] == [
{"id": department.id, "name": department.name}
]
def test_scoped_policy_forces_department_head_as_responsible(
user_factory,
department_factory,
):
department_head = user_factory(
"equipment-head@example.com",
phone_number="+79990001001",
)
user = user_factory(
"equipment-scoped@example.com",
phone_number="+79990001002",
)
department = department_factory("Эксплуатация", head=department_head)
_grant_equipment_role(user=user, department=department)
attributes = EquipmentCreationPolicy.resolve(user).validate_create(
{"department": department, "responsible_person": user}
)
assert attributes["responsible_person"] == department_head
def test_responsible_must_belong_to_equipment_department(
user_factory,
department_factory,
):
responsible = user_factory("equipment-outsider@example.com")
department = department_factory("Склад")
with pytest.raises(EquipmentPolicyValidationError) as exc_info:
validate_responsible_department(
department=department,
responsible=responsible,
)
assert "responsible_person" in exc_info.value.errors

156
backend/tests/api/v1/procurement/test_request_application.py Normal file
View File

@ -0,0 +1,156 @@
from decimal import Decimal
import pytest
from procurement.application import (
ProcurementRequestItemNotFound,
create_procurement_request,
update_procurement_request_with_items,
)
from procurement.constants import ProcurementStatus, UrgencyLevel
from procurement.models import ProcurementItem, ProcurementRequest
pytestmark = pytest.mark.django_db
def test_create_request_use_case_persists_items_and_initial_comments_atomically(
user_factory,
department_factory,
monkeypatch,
):
user = user_factory("request-app-create@example.com")
customer_department = department_factory("Заказчик")
processing_department = department_factory("Исполнитель")
comments = []
monkeypatch.setattr(
"procurement.application.requests.comments_helpers.create_comment",
lambda **kwargs: comments.append(kwargs),
)
procurement_request = create_procurement_request(
requestor=user,
attributes={
"title": "Расходные материалы",
"department": customer_department,
"processing_department": processing_department,
"urgency": UrgencyLevel.MEDIUM,
},
items=[
{
"name": "Перчатки",
"quantity": 2,
"unit": "упак",
"estimated_unit_price": Decimal("500.00"),
"initial_comment": "Срочная потребность",
}
],
)
assert procurement_request.status == ProcurementStatus.WAITING
item = procurement_request.items.get()
assert item.name == "Перчатки"
assert comments == [
{
"obj": item,
"author": user,
"content": "Срочная потребность",
}
]
def test_update_request_use_case_synchronizes_item_collection(
user_factory,
department_factory,
procurement_request_factory,
procurement_item_factory,
):
user = user_factory("request-app-update@example.com")
department = department_factory("Закупки")
procurement_request = procurement_request_factory(department, user)
retained_item = procurement_item_factory(
procurement_request,
name="Старое название",
)
obsolete_item = procurement_item_factory(
procurement_request,
name="Удалить",
)
updated = update_procurement_request_with_items(
request_id=procurement_request.id,
attributes={"title": "Обновлённая заявка"},
items=[
{
"id": retained_item.id,
"name": "Новое название",
"description": "",
"quantity": 3,
"unit": "шт",
"estimated_unit_price": Decimal("20.00"),
"supplier_info": "",
"links": [],
},
{
"name": "Новая позиция",
"description": "",
"quantity": 1,
"unit": "шт",
"estimated_unit_price": None,
"supplier_info": "",
"links": [],
},
],
author=user,
)
assert updated.title == "Обновлённая заявка"
assert set(updated.items.values_list("name", flat=True)) == {
"Новое название",
"Новая позиция",
}
assert not ProcurementItem.objects.filter(pk=obsolete_item.id).exists()
def test_update_request_use_case_rolls_back_when_item_is_foreign(
user_factory,
department_factory,
procurement_request_factory,
procurement_item_factory,
):
user = user_factory("request-app-rollback@example.com")
department = department_factory("Снабжение")
target_request = procurement_request_factory(
department,
user,
title="Исходное название",
)
other_request = procurement_request_factory(
department,
user,
title="Другая заявка",
)
foreign_item = procurement_item_factory(other_request)
with pytest.raises(ProcurementRequestItemNotFound):
update_procurement_request_with_items(
request_id=target_request.id,
attributes={"title": "Не должно сохраниться"},
items=[
{
"id": foreign_item.id,
"name": foreign_item.name,
"description": foreign_item.description,
"quantity": foreign_item.quantity,
"unit": foreign_item.unit,
"estimated_unit_price": foreign_item.estimated_unit_price,
"supplier_info": foreign_item.supplier_info,
"links": foreign_item.links,
}
],
author=user,
)
target_request.refresh_from_db()
assert target_request.title == "Исходное название"
assert ProcurementRequest.objects.filter(pk=other_request.id).exists()