api/src/api/v1/forms.py

@@ -11,7 +11,7 @@ from src.db.models.form_type import FormTypeEnum
 from src.db.models.role import UserRoleEnum
 from src.services.sheet_service import SheetService
 from src.api.v1.json_response import list_json_response
-from src.domain.schemas import AddLineSchema, BaseListResponse, BaseSingleResponse, ResponseBase, BudgetFormResponse, CellPatch, CellsPatch, DirectionSchemaEnum, FormCreateSchema, SheetFormTypeResponse, SheetResponse
+from src.domain.schemas import AddLineSchema, BaseListResponse, BaseSingleResponse, ResponseBase, BudgetFormResponse, CellFormulaResponse, CellPatch, CellsPatch, DirectionSchemaEnum, FormCreateSchema, SheetFormTypeResponse, SheetResponse
 from src.db.models.app_user import AppUser
 from src.services.budget_form_service import BudgetFormService
 from src.api.v1.deps import get_current_active_user_with_set_db, require_admin
@@ -236,6 +236,7 @@ async def update_cell(
         line_id=cell_body.line_id, 
         column=cell_body.column, 
         value=cell_body.value, 
+        formula=cell_body.formula,
         user=current_user,
     )
     db_ms = (time.perf_counter() - t0) * 1000
@@ -264,6 +265,26 @@ async def update_cell(
 
 
 
+@router.get("/{form_id}/sheet/{sheet}/formulas")
+async def get_sheet_formulas(
+        form_id: int,
+        sheet: str,
+        db: AsyncSession = Depends(get_db),
+        current_user: AppUser = Depends(get_current_active_user_with_set_db),
+) -> BaseListResponse[CellFormulaResponse]:
+    form = await BudgetFormService(db).get(
+        user=current_user,
+        budget_form_id=form_id,
+        load_form_type=True,
+    )
+    if not form:
+        raise HTTPException(404, "Форма не найдена")
+    if sheet not in form.form_type.sheet_list:
+        raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
+    result = await SheetService(db).list_formulas(form_id=form_id, sheet=sheet)
+    return BaseListResponse(count=len(result), result=result)
+
+
 @router.patch("/{form_id}/sheet/{sheet}/cells")
 async def update_cells(
         form_id: int, 
@@ -298,7 +319,10 @@ async def update_cells(
         sheet=sheet,
         direction=direction.value if direction else None,
         sections=sections,
-        changes=[c.model_dump() for c in cells_body.changes],
+        changes=[
+            {**c.model_dump(exclude={"formula"}), **({"formula": c.formula} if c.formula is not None else {})}
+            for c in cells_body.changes
+        ],
         user=current_user,
     )
     db_ms = (time.perf_counter() - t0) * 1000

api/src/core/exception_handlers.py

@@ -1,3 +1,5 @@
+import re
+
 from fastapi import FastAPI, HTTPException, status
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
@@ -28,6 +30,14 @@ def _error_payload(code: int, message: str, field: str = "") -> dict:
     }
 
 
+_ASYNCPG_CLASS_PREFIX = re.compile(r"^<class '[^']+'>:\s*")
+
+
+def _strip_driver_class_prefix(message: str) -> str:
+    """asyncpg-диалект даёт str(orig) вида "<class '...RaiseError'>: msg"; убираем обёртку."""
+    return _ASYNCPG_CLASS_PREFIX.sub("", message)
+
+
 def _parse_prefixed_validation_error(message: str) -> str:
     for prefix in VALIDATION_PREFIXES:
         if message.lower().startswith(prefix):
@@ -51,10 +61,11 @@ def map_sqlalchemy_error(exc: SQLAlchemyError) -> tuple[int, int, str, str]:
     message = _first_db_message_line(raw) or "Database error"
     low = message.lower()
 
-    if any(low.startswith(prefix) for prefix in VALIDATION_PREFIXES):
+    bare_message = _strip_driver_class_prefix(message)
+    if any(bare_message.lower().startswith(prefix) for prefix in VALIDATION_PREFIXES):
         status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
-        field = _parse_prefixed_validation_error(message)
-        return status_code, status_code, message, field
+        field = _parse_prefixed_validation_error(bare_message)
+        return status_code, status_code, bare_message, field
     if any(error in low for error in PG_404_ERRORS):
         status_code = status.HTTP_404_NOT_FOUND
         return status_code, status_code, message, ""

api/src/services/sheet_service.py

@@ -8,6 +8,7 @@ from src.repository.form4_project_repository import Form4ProjectRepository
 from src.repository.user_repository import UserRepository
 from src.db.models.app_user import AppUser
 from src.db.models.budget_form import BudgetForm
+from src.repository.cell_formula_repository import CellFormulaRepository
 from src.repository.sheet_repository import SheetRepository, SheetValidationError
 
 
@@ -16,6 +17,7 @@ class SheetService:
     def __init__(self, db: AsyncSession):
         self.db = db
         self.sheet_repo = SheetRepository(db)
+        self.formula_repo = CellFormulaRepository(db)
         self.bf_repo = BudgetFormRepository(db)
         self.user_repo = UserRepository(db)
         self.f4_project_repo = Form4ProjectRepository(db)
@@ -61,7 +63,7 @@ class SheetService:
             changes: list[dict],
             user: AppUser | None = None,
     ) -> list[tuple]:
-        return await self.sheet_repo.update_cells(
+        rows = await self.sheet_repo.update_cells(
             form_id=form_id, 
             sheet=sheet,
             direction=direction,
@@ -69,28 +71,81 @@ class SheetService:
             changes=changes,
             user_id=user.id if user else None,
         )
-    
-    async def update_cell(
-            self, 
-            form_id: int, 
+        if user:
+            await self.formula_repo.set_formulas(
+                budget_form_id=form_id,
+                report_id=None,
+                sheet=sheet,
+                changes=changes,
+                user_id=user.id,
+            )
+        return rows
+
+    async def update_cell_with_formula(
+            self,
+            form_id: int,
             sheet: str,
             direction: str | None,
             sections: list[str] | None,
-            line_id: int, 
-            column: str, 
-            value: str | int | float | None = None, 
+            line_id: int,
+            column: str,
+            value: str | int | float | None = None,
+            formula: str | None = None,
             user: AppUser | None = None,
-    ) -> list[tuple]:
-        return await self.sheet_repo.update_cell(
-            form_id=form_id, 
+    ) -> tuple[list[tuple], dict | None]:
+        rows = await self.sheet_repo.update_cell(
+            form_id=form_id,
             sheet=sheet,
             direction=direction,
             sections=sections,
-            line_id=line_id, 
-            column=column, 
-            value=value, 
+            line_id=line_id,
+            column=column,
+            value=value,
             user_id=user.id if user else None,
         )
+        if not user:
+            return rows, None
+        change = {"line_id": line_id, "column": column, "value": value}
+        if formula is not None:
+            change["formula"] = formula
+        saved = await self.formula_repo.set_formulas(
+            budget_form_id=form_id,
+            report_id=None,
+            sheet=sheet,
+            changes=[change],
+            user_id=user.id,
+        )
+        return rows, (saved[0] if saved else None)
+
+    async def update_cell(
+            self,
+            form_id: int,
+            sheet: str,
+            direction: str | None,
+            sections: list[str] | None,
+            line_id: int,
+            column: str,
+            value: str | int | float | None = None,
+            formula: str | None = None,
+            user: AppUser | None = None,
+    ) -> list[tuple]:
+        rows, _ = await self.update_cell_with_formula(
+            form_id=form_id,
+            sheet=sheet,
+            direction=direction,
+            sections=sections,
+            line_id=line_id,
+            column=column,
+            value=value,
+            formula=formula,
+            user=user,
+        )
+        return rows
+
+    async def list_formulas(self, form_id: int, sheet: str) -> list[dict]:
+        return await self.formula_repo.list_formulas(
+            budget_form_id=form_id, report_id=None, sheet=sheet
+        )
 
 
     async def add_line(

api/tests/integration/test_cell_formula_integration.py

@@ -0,0 +1,177 @@
+"""AURORA-1896: формулы ячеек форм (REST): запись при PATCH и GET formulas."""
+import pytest
+
+FORM = "/api/v1/form/1"
+QS = "?direction=Support&sections=plan,contract_summary"
+
+
+@pytest.fixture
+def headers(admin_tokens, auth_headers):
+    return auth_headers(admin_tokens)
+
+
+def _add_line(client, headers) -> int:
+    r = client.post(
+        f"{FORM}/sheet/AHR/line{QS}",
+        json={"expense_item_id": 120, "direction": "Support"},
+        headers=headers,
+    )
+    assert r.status_code == 200, r.text
+    inputs = [x for x in r.json()["result"] if x.get("row_type") == "INPUT"]
+    assert len(inputs) == 1
+    return inputs[0]["data"]["line_id"]
+
+
+def _patch(client, headers, body, path="cell"):
+    return client.patch(f"{FORM}/sheet/AHR/{path}{QS}", json=body, headers=headers)
+
+
+def _formulas(client, headers) -> list[dict]:
+    r = client.get(f"{FORM}/sheet/AHR/formulas", headers=headers)
+    assert r.status_code == 200, r.text
+    return r.json()["result"]
+
+
+def _cell_value(client, headers, line_id, column="plan.q1"):
+    r = client.get(f"{FORM}/sheet/AHR{QS}", headers=headers)
+    assert r.status_code == 200, r.text
+    for row in r.json()["result"]:
+        if row.get("row_type") == "INPUT" and row["data"].get("line_id") == line_id:
+            node = row["data"]
+            for part in column.split("."):
+                node = node.get(part) if isinstance(node, dict) else None
+            return node
+    return "ROW_NOT_FOUND"
+
+
+def _line_formulas(client, headers, line_id):
+    return [f for f in _formulas(client, headers) if f["line_id"] == line_id]
+
+
+def test_patch_cell_with_formula_then_get(client, headers, admin_tokens):
+    line_id = _add_line(client, headers)
+    r = _patch(client, headers, {"line_id": line_id, "column": "plan.q1", "value": 150, "formula": " =100+50 "})
+    assert r.status_code == 200, r.text
+    found = _line_formulas(client, headers, line_id)
+    assert len(found) == 1
+    f = found[0]
+    assert f["column"] == "plan.q1"
+    assert f["expression"] == "100+50"
+    assert f["value"] == 150
+    me = client.get("/api/v1/auth/me", headers=headers)
+    if me.status_code == 200:
+        assert f["updated_by"]["id"] == me.json().get("id", f["updated_by"]["id"])
+    assert f["updated_by"]["id"] > 0
+    assert f["updated_at"]
+
+
+def test_patch_without_formula_removes(client, headers):
+    line_id = _add_line(client, headers)
+    assert _patch(client, headers, {"line_id": line_id, "column": "plan.q1", "value": 150, "formula": "100+50"}).status_code == 200
+    assert len(_line_formulas(client, headers, line_id)) == 1
+    assert _patch(client, headers, {"line_id": line_id, "column": "plan.q1", "value": 200}).status_code == 200
+    assert _line_formulas(client, headers, line_id) == []
+
+
+def test_patch_invalid_formula_422_value_unchanged(client, headers):
+    before = _cell_value(client, headers, 1)
+    r = _patch(client, headers, {"line_id": 1, "column": "plan.q1", "value": 777, "formula": "drop table"})
+    assert r.status_code == 422, r.text
+    assert "invalid_formula" in r.text
+    assert r.json()["field"] == "chars"
+    # сессия теста откатывается целиком (conftest) — GET ниже видит состояние после отката
+    assert _cell_value(client, headers, 1) == before
+    assert _line_formulas(client, headers, 1) == []
+
+
+@pytest.mark.parametrize(
+    "line_col,value,formula",
+    [("plan.q1", 777, "drop table"), ("plan.q1", None, "1+1"), ("header.justification", "текст", "1+1")],
+    ids=["bad-chars", "null-value", "text-column"],
+)
+def test_invalid_formula_does_not_write_value(client, headers, line_col, value, formula):
+    """Атомарность без опоры на откат фикстуры: ошибку ловим внутри SAVEPOINT."""
+    from sqlalchemy.exc import DBAPIError
+
+    from src.db.models.app_user import AppUser
+    from src.services.sheet_service import SheetService
+
+    line_id = _add_line(client, headers)
+    before = _cell_value(client, headers, line_id, line_col)
+
+    async def _run():
+        db = client.db_session
+        user = await db.get(AppUser, 1)
+        try:
+            async with db.begin_nested():
+                await SheetService(db).update_cell(
+                    form_id=1, sheet="AHR", direction="Support", sections=["plan", "contract_summary"],
+                    line_id=line_id, column=line_col, value=value, formula=formula, user=user,
+                )
+        except DBAPIError as exc:
+            return str(exc.orig)
+        return None
+
+    err = client.portal.call(_run)
+    assert err is not None and "invalid_formula" in err
+    assert _cell_value(client, headers, line_id, line_col) == before
+    assert _line_formulas(client, headers, line_id) == []
+
+
+def test_patch_formula_into_text_column_422(client, headers):
+    line_id = _add_line(client, headers)
+    r = _patch(client, headers, {"line_id": line_id, "column": "header.justification", "value": "текст", "formula": "1+1"})
+    assert r.status_code == 422, r.text
+    assert "invalid_formula" in r.text
+    assert _line_formulas(client, headers, line_id) == []
+
+
+def test_patch_formula_null_value_422(client, headers):
+    line_id = _add_line(client, headers)
+    r = _patch(client, headers, {"line_id": line_id, "column": "plan.q1", "value": None, "formula": "1+1"})
+    assert r.status_code == 422, r.text
+    assert "invalid_formula" in r.text
+    assert _line_formulas(client, headers, line_id) == []
+
+
+def test_patch_formula_forbidden_403_no_formula(client, headers, isp2_tokens, auth_headers):
+    # isp2 в оргединице формы 1, но колонку plan.q2 ему править нельзя (role_not_allowed).
+    # isp1 форму 1 не видит вовсе (404), поэтому для 403 берём isp2.
+    r = client.patch(
+        f"{FORM}/sheet/AHR/cell{QS}",
+        json={"line_id": 1, "column": "plan.q2", "value": 5, "formula": "2+3"},
+        headers=auth_headers(isp2_tokens),
+    )
+    assert r.status_code == 403, r.text
+    assert [f for f in _formulas(client, headers) if f["line_id"] == 1] == []
+
+
+def test_patch_cells_mixed(client, headers):
+    l1 = _add_line(client, headers)
+    l2 = _add_line(client, headers)
+    r = _patch(
+        client,
+        headers,
+        {"changes": [
+            {"line_id": l1, "column": "plan.q1", "value": 3, "formula": "1+2"},
+            {"line_id": l2, "column": "plan.q1", "value": 4},
+        ]},
+        path="cells",
+    )
+    assert r.status_code == 200, r.text
+    assert [f["expression"] for f in _line_formulas(client, headers, l1)] == ["1+2"]
+    assert _line_formulas(client, headers, l2) == []
+
+
+def test_delete_line_removes_formulas(client, headers):
+    line_id = _add_line(client, headers)
+    assert _patch(client, headers, {"line_id": line_id, "column": "plan.q1", "value": 3, "formula": "1+2"}).status_code == 200
+    assert len(_line_formulas(client, headers, line_id)) == 1
+    r = client.delete(f"{FORM}/sheet/AHR/line/{line_id}?direction=Support", headers=headers)
+    assert r.status_code == 200, r.text
+    assert _line_formulas(client, headers, line_id) == []
+
+
+def test_get_formulas_unknown_form_404(client, headers):
+    r = client.get("/api/v1/form/999/sheet/AHR/formulas", headers=headers)
+    assert r.status_code == 404