api/src/core/CONSTANTS.py

@@ -8,6 +8,7 @@ VALIDATION_PREFIXES = (
     "bad_booking_year:",
     "bad_booking_quarter:",
     "unsupported_scope:",
+    "invalid_formula:",
 )
 
 PG_404_ERRORS = (

api/src/domain/schemas.py

@@ -394,10 +394,25 @@ class Svod2Facets(BaseModel):
     org_name: list[Svod2FacetValue]
 
 
+class UserShort(BaseModel):
+    id: int
+    full_name: str | None = None
+
+
+class CellFormulaResponse(BaseModel):
+    line_id: int
+    column: str
+    expression: str
+    value: float
+    updated_by: UserShort
+    updated_at: datetime
+
+
 class CellPatch(BaseModel):
     line_id: int
     column: str
     value: Any = None
+    formula: str | None = None
 
 
 class CellsPatch(BaseModel):

api/src/repository/cell_formula_repository.py

@@ -0,0 +1,65 @@
+import json
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+
+def _map_row(row) -> dict:
+    line_id, column_key, expression, value, user_id, user_name, updated_at = tuple(row)
+    return {
+        "line_id": line_id,
+        "column": column_key,
+        "expression": expression,
+        "value": float(value),
+        "updated_by": {"id": user_id, "full_name": user_name},
+        "updated_at": updated_at,
+    }
+
+
+class CellFormulaRepository:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def set_formulas(
+        self,
+        *,
+        budget_form_id: int | None,
+        report_id: int | None,
+        sheet: str,
+        changes: list[dict],
+        user_id: int,
+    ) -> list[dict]:
+        result = await self.db.execute(
+            text(
+                """
+                SELECT line_id, column_key, expression, value, updated_by, updated_by_name, updated_at
+                FROM v3.set_cell_formulas(:budget_form_id, :report_id, :sheet, CAST(:changes AS JSONB), :user_id)
+                """
+            ),
+            {
+                "budget_form_id": budget_form_id,
+                "report_id": report_id,
+                "sheet": sheet,
+                "changes": json.dumps(changes),
+                "user_id": user_id,
+            },
+        )
+        return [_map_row(r) for r in result.all()]
+
+    async def list_formulas(
+        self,
+        *,
+        budget_form_id: int | None,
+        report_id: int | None,
+        sheet: str,
+    ) -> list[dict]:
+        result = await self.db.execute(
+            text(
+                """
+                SELECT line_id, column_key, expression, value, updated_by, updated_by_name, updated_at
+                FROM v3.v_cell_formulas(:budget_form_id, :report_id, :sheet)
+                """
+            ),
+            {"budget_form_id": budget_form_id, "report_id": report_id, "sheet": sheet},
+        )
+        return [_map_row(r) for r in result.all()]

api/tests/unit/test_cell_formula_repository.py

@@ -0,0 +1,88 @@
+import json
+from datetime import datetime, timezone
+from decimal import Decimal
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.repository.cell_formula_repository import CellFormulaRepository
+
+DT = datetime(2026, 11, 11, 10, 30, tzinfo=timezone.utc)
+ROW = (10, "plan.q1", "100+50", Decimal("150"), 7, "Иванов И.И.", DT)
+EXPECTED = {
+    "line_id": 10,
+    "column": "plan.q1",
+    "expression": "100+50",
+    "value": 150.0,
+    "updated_by": {"id": 7, "full_name": "Иванов И.И."},
+    "updated_at": DT,
+}
+
+
+def _result(rows):
+    result = MagicMock()
+    result.all.return_value = rows
+    return result
+
+
+@pytest.fixture
+def mock_db():
+    return AsyncMock(spec=AsyncSession)
+
+
+@pytest.fixture
+def repository(mock_db):
+    return CellFormulaRepository(mock_db)
+
+
+@pytest.mark.asyncio
+async def test_set_formulas_calls_sql_with_json_changes(repository, mock_db):
+    mock_db.execute.return_value = _result([])
+    changes = [{"line_id": 10, "column": "plan.q1", "value": 150, "formula": "=100+50"}]
+
+    await repository.set_formulas(
+        budget_form_id=1, report_id=None, sheet="AHR", changes=changes, user_id=7
+    )
+
+    mock_db.execute.assert_awaited_once()
+    stmt, params = mock_db.execute.await_args.args
+    assert "v3.set_cell_formulas" in str(stmt)
+    assert params["budget_form_id"] == 1
+    assert params["report_id"] is None
+    assert params["sheet"] == "AHR"
+    assert params["user_id"] == 7
+    assert isinstance(params["changes"], str)
+    assert json.loads(params["changes"])[0]["formula"] == "=100+50"
+
+
+@pytest.mark.asyncio
+async def test_set_formulas_maps_rows(repository, mock_db):
+    mock_db.execute.return_value = _result([ROW])
+
+    result = await repository.set_formulas(
+        budget_form_id=1, report_id=None, sheet="AHR", changes=[], user_id=7
+    )
+
+    assert result == [EXPECTED]
+
+
+@pytest.mark.asyncio
+async def test_list_formulas_calls_view(repository, mock_db):
+    mock_db.execute.return_value = _result([ROW])
+
+    result = await repository.list_formulas(budget_form_id=None, report_id=5, sheet="AHR")
+
+    stmt, params = mock_db.execute.await_args.args
+    assert "v3.v_cell_formulas" in str(stmt)
+    assert params == {"budget_form_id": None, "report_id": 5, "sheet": "AHR"}
+    assert result == [EXPECTED]
+
+
+@pytest.mark.asyncio
+async def test_mapping_handles_missing_user_name(repository, mock_db):
+    mock_db.execute.return_value = _result([(1, "q", "1+1", Decimal("2"), 3, None, DT)])
+
+    result = await repository.list_formulas(budget_form_id=1, report_id=None, sheet="CAP")
+
+    assert result[0]["updated_by"] == {"id": 3, "full_name": None}

api/tests/unit/test_exception_handlers.py

@@ -157,3 +157,15 @@ def test_map_http_exception_not_found_maps_code_from_status():
     assert code == status.HTTP_404_NOT_FOUND
     assert message == "not found"
     assert field == ""
+
+
+def test_map_sqlalchemy_error_invalid_formula_prefix():
+    exc = SQLAlchemyError()
+    exc.orig = Exception("invalid_formula: expression")
+
+    status_code, code, message, field = map_sqlalchemy_error(exc)
+
+    assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
+    assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
+    assert message == "invalid_formula: expression"
+    assert field == "expression"