RSHB/DFiP_Budget_planing
10
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(formulas): формулы в отчётах проектов — запись и GET formulas (AURORA-1896)

Browse Source 
PATCH cell/cells отчёта проекта пишут формулу в той же транзакции, GET
.../report/{year}/{report_type}/formulas возвращает формулы отчёта.
Тесты: roundtrip, снятие формулы, каскад удаления проекта, триггер
rf_project_report_line (del_form3_line и прямой DELETE).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-11-11 19:28:55 +04:00
parent f707130ff0
commit 187988bd76
3 changed files with 298 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
api/src/api/v1/projects.py
View File

@ -17,6 +17,7 @@ from src.domain.schemas import (
AddProjectBody,
BaseListResponse,
BaseSingleResponse,
CellFormulaResponse,
CellPatch,
CellsPatch,
ResponseBase,
@ -346,11 +347,26 @@ async def upd_form3_cell(
line_id=body.line_id,
column=body.column,
value=body.value,
formula=body.formula,
user=current_user,
)
return _rows_to_payload(rows)
@router.get("/projects/{project_id}/report/{year}/{report_type}/formulas")
async def get_form3_formulas(
project_id: int,
year: int,
report_type: str,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
) -> BaseListResponse[CellFormulaResponse]:
result = await ProjectService(db).list_formulas(
project_id=project_id, year=year, report_type=report_type, user=current_user
)
return BaseListResponse(count=len(result), result=result)
@router.patch("/projects/{project_id}/report/{year}/{report_type}/cells")
async def upd_form3_cells(
project_id: int,
@ -361,11 +377,17 @@ async def upd_form3_cells(
current_user: AppUser = Depends(get_current_active_user_with_set_db),
) -> list[SheetResponse]:
project_service = ProjectService(db)
changes = []
for change in body.changes:
item = change.model_dump(exclude={"formula"})
if change.formula is not None:
item["formula"] = change.formula
changes.append(item)
rows = await project_service.upd_form3_cells(
project_id=project_id,
year=year,
report_type=report_type,
changes=[change.model_dump() for change in body.changes],
changes=changes,
user=current_user,
)
return _rows_to_payload(rows)

69
api/src/services/project_service.py
View File

@ -5,6 +5,7 @@ from src.db.models.app_user import AppUser
from src.db.models.role import UserRoleEnum
from src.core.errors import AccessDeniedException, ValidationException
from src.domain.schemas import placement_type_to_db
from src.repository.cell_formula_repository import CellFormulaRepository
from src.repository.project_repository import ProjectRepository
from src.repository.user_repository import UserRepository
@ -14,6 +15,7 @@ class ProjectService:
self.db = db
self.project_repo = ProjectRepository(db)
self.user_repo = UserRepository(db)
self.formula_repo = CellFormulaRepository(db)
async def get_list(
self,
@ -282,6 +284,33 @@ class ProjectService:
project_type=project_type,
)
async def upd_form3_cell_with_formula(
self,
project_id: int,
year: int,
report_type: str,
line_id: int,
column: str,
value,
formula: str | None,
user: AppUser,
) -> tuple[list[tuple], dict | None]:
report_id = await self.resolve_report_id(project_id, year, report_type, user)
if not report_id:
raise ValidationException("Отчёт не найден")
rows = await self.project_repo.upd_form3_cell(report_id, line_id, column, value)
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=None,
report_id=report_id,
sheet=report_type.upper(),
changes=[change],
user_id=user.id,
)
return rows, (saved[0] if saved else None)
async def upd_form3_cell(
self,
project_id: int,
@ -291,11 +320,19 @@ class ProjectService:
column: str,
value,
user: AppUser,
formula: str | None = None,
) -> list[tuple]:
report_id = await self.resolve_report_id(project_id, year, report_type, user)
if not report_id:
raise ValidationException("Отчёт не найден")
return await self.project_repo.upd_form3_cell(report_id, line_id, column, value)
rows, _ = await self.upd_form3_cell_with_formula(
project_id=project_id,
year=year,
report_type=report_type,
line_id=line_id,
column=column,
value=value,
formula=formula,
user=user,
)
return rows
async def upd_form3_cells(
self,
@ -308,7 +345,29 @@ class ProjectService:
report_id = await self.resolve_report_id(project_id, year, report_type, user)
if not report_id:
raise ValidationException("Отчёт не найден")
return await self.project_repo.upd_form3_cells(report_id, changes)
rows = await self.project_repo.upd_form3_cells(report_id, changes)
await self.formula_repo.set_formulas(
budget_form_id=None,
report_id=report_id,
sheet=report_type.upper(),
changes=changes,
user_id=user.id,
)
return rows
async def list_formulas(
self,
project_id: int,
year: int,
report_type: str,
user: AppUser,
) -> list[dict]:
report_id = await self.resolve_report_id(project_id, year, report_type, user)
if not report_id:
raise ValidationException("Отчёт не найден")
return await self.formula_repo.list_formulas(
budget_form_id=None, report_id=report_id, sheet=report_type.upper()
)
async def add_form3_line(
self,

211
api/tests/integration/test_cell_formula_integration.py
View File

@ -175,3 +175,214 @@ def test_delete_line_removes_formulas(client, headers):
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
# ---- Отчёты проектов (FORM_3) ----
import uuid
from datetime import datetime
from sqlalchemy import text
def _create_project(client, headers, year) -> int:
r = client.post(
"/api/v1/projects",
json={
"name": f"PF_{uuid.uuid4().hex[:10]}",
"year": year,
"branch_id": 1,
"project_type": "Открытие ВСП",
"vsp_format": "Типовой",
"placement_type": "Собственность",
"object_address": "Тестовая улица 1",
},
headers=headers,
)
assert r.status_code == 200, r.text
return r.json()["project_id"]
def _add_project_line(client, headers, project_id, year, report_type="LIMIT", nth=0) -> int:
"""Строки отчёта создаются вместе с проектом: берём существующую nth-ю строку отчёта."""
async def _run():
res = await client.db_session.execute(
text(
"SELECT l.id FROM v3.rf_project_report_line l "
"JOIN v3.rf_project_report r ON r.id = l.rf_project_report_id "
"JOIN v3.rf_project_year py ON py.id = r.rf_project_year_id "
"WHERE r.project_id = :pid AND r.report_type = :rt AND py.year = :y "
"ORDER BY l.id LIMIT 1 OFFSET :n"
),
{"pid": project_id, "rt": report_type, "y": year, "n": nth},
)
return res.scalar()
line_id = client.portal.call(_run)
assert line_id, "у нового проекта нет строк отчёта"
return int(line_id)
def _report_url(project_id, year, report_type="LIMIT"):
return f"/api/v1/projects/{project_id}/report/{year}/{report_type}"
def _project_formulas(client, headers, project_id, year, report_type="LIMIT") -> list[dict]:
r = client.get(f"{_report_url(project_id, year, report_type)}/formulas", headers=headers)
assert r.status_code == 200, r.text
return r.json()["result"]
def _scalar(client, sql, params):
async def _run():
return (await client.db_session.execute(text(sql), params)).scalar()
return client.portal.call(_run)
def test_project_report_formula_roundtrip(client, headers):
year = datetime.now().year
pid = _create_project(client, headers, year)
line_id = _add_project_line(client, headers, pid, year)
r = client.patch(
f"{_report_url(pid, year)}/cell",
json={"line_id": line_id, "column": "q1.m1", "value": 150, "formula": " =100+50 "},
headers=headers,
)
assert r.status_code == 200, r.text
found = _project_formulas(client, headers, pid, year)
assert len(found) == 1
assert found[0]["line_id"] == line_id
assert found[0]["column"] == "q1.m1"
assert found[0]["expression"] == "100+50"
assert found[0]["value"] == 150
assert found[0]["updated_by"]["id"] > 0
# нижний регистр report_type в пути
assert _project_formulas(client, headers, pid, year, "limit") == found
# другой тип отчёта формулу не видит
assert _project_formulas(client, headers, pid, year, "CURRENT_EXPENSES") == []
def test_project_report_formula_removed_without_formula(client, headers):
year = datetime.now().year
pid = _create_project(client, headers, year)
line_id = _add_project_line(client, headers, pid, year)
url = f"{_report_url(pid, year)}/cell"
assert client.patch(url, json={"line_id": line_id, "column": "q1.m1", "value": 150, "formula": "100+50"}, headers=headers).status_code == 200
assert len(_project_formulas(client, headers, pid, year)) == 1
assert client.patch(url, json={"line_id": line_id, "column": "q1.m1", "value": 200}, headers=headers).status_code == 200
assert _project_formulas(client, headers, pid, year) == []
def test_project_report_cells_formulas(client, headers):
year = datetime.now().year
pid = _create_project(client, headers, year)
line_id = _add_project_line(client, headers, pid, year)
r = client.patch(
f"{_report_url(pid, year)}/cells",
json={"changes": [
{"line_id": line_id, "column": "q1.m1", "value": 3, "formula": "1+2"},
{"line_id": line_id, "column": "q1.m2", "value": 5},
]},
headers=headers,
)
assert r.status_code == 200, r.text
found = _project_formulas(client, headers, pid, year)
assert [(f["column"], f["expression"]) for f in found] == [("q1.m1", "1+2")]
def test_project_report_invalid_formula_422(client, headers):
year = datetime.now().year
pid = _create_project(client, headers, year)
line_id = _add_project_line(client, headers, pid, year)
r = client.patch(
f"{_report_url(pid, year)}/cell",
json={"line_id": line_id, "column": "q1.m1", "value": 5, "formula": "drop table"},
headers=headers,
)
assert r.status_code == 422, r.text
assert "invalid_formula" in r.text
def test_project_formulas_report_not_found(client, headers):
pid = _create_project(client, headers, datetime.now().year)
r = client.get(f"{_report_url(pid, 2099)}/formulas", headers=headers)
assert r.status_code == 400
assert r.json().get("message") == "Отчёт не найден"
def test_project_delete_cascades_formulas(client, headers):
year = datetime.now().year
pid = _create_project(client, headers, year)
line_id = _add_project_line(client, headers, pid, year)
assert client.patch(
f"{_report_url(pid, year)}/cell",
json={"line_id": line_id, "column": "q1.m1", "value": 3, "formula": "1+2"},
headers=headers,
).status_code == 200
count_sql = (
"SELECT count(*) FROM v3.cell_formula WHERE rf_project_report_id IN "
"(SELECT id FROM v3.rf_project_report WHERE project_id = :pid)"
)
assert _scalar(client, count_sql, {"pid": pid}) == 1
report_ids = client.portal.call(
lambda: _report_ids(client, pid)
)
r = client.delete(f"/api/v1/projects/{pid}", headers=headers)
assert r.status_code == 200, r.text
assert report_ids
cnt = _scalar(
client,
"SELECT count(*) FROM v3.cell_formula WHERE rf_project_report_id = ANY(:ids)",
{"ids": report_ids},
)
assert cnt == 0
async def _report_ids(client, pid):
res = await client.db_session.execute(
text("SELECT id FROM v3.rf_project_report WHERE project_id = :pid"), {"pid": pid}
)
return [row[0] for row in res.all()]
def test_del_form3_line_removes_formula(client, headers):
year = datetime.now().year
pid = _create_project(client, headers, year)
line_id = _add_project_line(client, headers, pid, year)
assert client.patch(
f"{_report_url(pid, year)}/cell",
json={"line_id": line_id, "column": "q1.m1", "value": 3, "formula": "1+2"},
headers=headers,
).status_code == 200
assert len(_project_formulas(client, headers, pid, year)) == 1
r = client.delete(f"{_report_url(pid, year)}/line/{line_id}", headers=headers)
assert r.status_code == 200, r.text
assert _project_formulas(client, headers, pid, year) == []
def test_rf_project_report_line_trigger_direct_delete(client, headers):
"""Триггер на rf_project_report_line: прямой DELETE строки чистит формулу
только этой строки (формула соседней строки того же отчёта остаётся)."""
year = datetime.now().year
pid = _create_project(client, headers, year)
l1 = _add_project_line(client, headers, pid, year)
l2 = _add_project_line(client, headers, pid, year, nth=1)
other = [l2]
for line in [l1] + other[:1]:
assert client.patch(
f"{_report_url(pid, year)}/cell",
json={"line_id": line, "column": "q1.m1", "value": 3, "formula": "1+2"},
headers=headers,
).status_code == 200
async def _delete_direct():
db = client.db_session
await db.execute(text("DELETE FROM v3.rf_project_report_quarter WHERE rf_project_report_line_id = :l"), {"l": l1})
await db.execute(text("DELETE FROM v3.rf_project_report_line WHERE id = :l"), {"l": l1})
client.portal.call(_delete_direct)
left = {f["line_id"] for f in _project_formulas(client, headers, pid, year)}
assert l1 not in left
if other:
assert other[0] in left