phase-template-edit: редактирование шаблонов этапов

api/src/api/v1/phase_template.py

@@ -0,0 +1,117 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.services.phase_template_service import PhaseTemplateService
+from src.db.models.app_user import AppUser
+from src.domain.schemas import (
+    BaseListResponse,
+    BaseSingleResponse,
+    FormTypeSchemaEnum,
+    PhaseTemplateCreate,
+    PhaseTemplateDelete,
+    PhaseTemplateResponse,
+    PhaseTemplateUpdate,
+)
+from src.api.v1.deps import require_admin
+from src.db.session import get_db
+
+router = APIRouter(prefix="/stages-templates", tags=["stages-templates"])
+
+
+@router.get("/")
+async def get_phase_templates(
+    form_type: FormTypeSchemaEnum | None = None,
+    sheet: str | None = None,
+    direction: str | None = None,
+    role: str | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+) -> BaseListResponse[PhaseTemplateResponse]:
+    pt_service = PhaseTemplateService(db)
+
+    templates = await pt_service.get_list(
+        form_type=form_type,
+        sheet=sheet,
+        direction=direction,
+        role=role,
+    )
+    return BaseListResponse(
+        result=[PhaseTemplateResponse.model_validate(t) for t in templates],
+        count=len(templates),
+    )
+
+
+@router.post("/")
+async def create_phase_template(
+    body: PhaseTemplateCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+) -> BaseSingleResponse[PhaseTemplateResponse]:
+    pt_service = PhaseTemplateService(db)
+    template = await pt_service.create(
+        form_type=body.form_type,
+        sheet=body.sheet,
+        phase_code=body.phase_code,
+        direction=body.direction,
+        role=body.role,
+        column_keys=body.column_keys,
+        opens_at=body.opens_at,
+        closes_at=body.closes_at,
+    )
+    return BaseSingleResponse(result=PhaseTemplateResponse.model_validate(template))
+
+
+@router.patch("/{form_type}/{sheet}/{phase_code}")
+async def update_phase_template(
+    form_type: FormTypeSchemaEnum,
+    sheet: str,
+    phase_code: str,
+    body: PhaseTemplateUpdate,
+    direction: str | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+) -> BaseSingleResponse[PhaseTemplateResponse]:
+    pt_service = PhaseTemplateService(db)
+    template = await pt_service.update(
+        form_type=form_type,
+        sheet=sheet,
+        phase_code=phase_code,
+        direction=direction,
+        data=body.model_dump(exclude_unset=True),
+    )
+    if not template:
+        raise HTTPException(404, "Шаблон этапа не найден")
+    return BaseSingleResponse(result=PhaseTemplateResponse.model_validate(template))
+
+
+@router.delete("/{form_type}/{sheet}/{phase_code}")
+async def delete_phase_template(
+    form_type: FormTypeSchemaEnum,
+    sheet: str,
+    phase_code: str,
+    direction: str | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+):
+    pt_service = PhaseTemplateService(db)
+    deleted = await pt_service.delete(
+        form_type=form_type,
+        sheet=sheet,
+        direction=direction,
+        phase_code=phase_code,
+    )
+    if not deleted:
+        raise HTTPException(404, "Шаблон этапа не найден")
+
+
+@router.delete("/bulk-delete")
+async def bulk_delete_phase_templates(
+    body: list[PhaseTemplateDelete],
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+) -> BaseSingleResponse[bool]:
+    pt_service = PhaseTemplateService(db)
+    deleted = await pt_service.bulk_delete(
+        lst=[item.model_dump() for item in body],
+    )
+    return BaseSingleResponse(result=deleted)

api/src/api/v1/router.py

@@ -1,7 +1,8 @@
 from fastapi import APIRouter
 
 from src.api.v1 import (
-    auth, users, admin, audit, forms, form_phases, projects, export, org_unit, vsp, expense_item, summary
+    auth, users, admin, audit, forms, form_phases, projects, export, org_unit, vsp, expense_item, summary,
+    phase_template,
 )
 
 from src.api.v1 import websocket
@@ -21,3 +22,4 @@ api_router.include_router(org_unit.router)
 api_router.include_router(vsp.router)
 api_router.include_router(expense_item.router)
 api_router.include_router(summary.router)
+api_router.include_router(phase_template.router)

api/src/db/models/form_type.py

@@ -38,7 +38,7 @@ class FormType(Base):
         FormTypeEnum.FORM_4: ["AHR", "CAP", "OPER", "STRUCTURE", "SMETA"],
     }
     SECTIONS_BY_FORM_TYPE = {
-        FormTypeEnum.FORM_1: ["plan", "contract_summary", "allocation", "sequestration", "reserve", "approved", "collegial", "ckk", "contract_detail", "q1", "q2", "q3", "q4", "totals"],
+        FormTypeEnum.FORM_1: ["header", "plan", "contract_summary", "allocation", "sequestration", "reserve", "approved", "collegial", "ckk", "contract_detail", "q1", "q2", "q3", "q4", "totals"],
         FormTypeEnum.FORM_2: ["header", "plan", "seq_dfip", "seq_ssp", "approved", "contract", "booking", "q1", "q2", "q3", "q4", "totals", "contract_summary", "allocation", "collegial", "ckk", "fact_q1", "fact_q2", "fact_q3", "fact_q4"],
         FormTypeEnum.FORM_3: ["q1", "q2", "q3", "q4", "year"],
         FormTypeEnum.FORM_4: ["header", "plan", "seq_dfip", "approved", "contract", "booking", "q1", "q2", "q3", "q4", "totals", "contract_summary", "allocation", "reserve", "collegial", "ckk"],

api/src/domain/schemas.py

@@ -458,6 +458,20 @@ class FormPhaseResponse(BaseModel):
     direction: str | None = None
 
 
+
+class PhaseTemplateResponse(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    form_type: FormTypeSchemaEnum
+    sheet: str
+    phase_code: str
+    role: FormPhaseRole
+    column_keys: list[str]
+    opens_at: str
+    closes_at: str
+    direction: str | None = None
+
+
 class FormPhaseCreate(BaseModel):
     sheet: str
     phase_code: str
@@ -468,6 +482,17 @@ class FormPhaseCreate(BaseModel):
     direction: str | None = None
 
 
+class PhaseTemplateCreate(BaseModel):
+    form_type: FormTypeSchemaEnum | str
+    sheet: str
+    phase_code: str
+    role: FormPhaseRole
+    column_keys: list[str]
+    opens_at: str
+    closes_at: str
+    direction: str | None = None
+
+
 class FormPhaseUpdate(BaseModel):
     role: FormPhaseRole | None = None
     column_keys: list[str] | None = None
@@ -475,6 +500,20 @@ class FormPhaseUpdate(BaseModel):
     closes_at: datetime | None = None
 
 
+class PhaseTemplateUpdate(BaseModel):
+    role: FormPhaseRole | None = None
+    column_keys: list[str] | None = None
+    opens_at: str | None = None
+    closes_at: str | None = None
+
+
+class PhaseTemplateDelete(BaseModel):
+    form_type: FormTypeSchemaEnum | str
+    sheet: str
+    phase_code: str
+    direction: str | None = None
+
+
 class Form3PhaseResponse(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 

api/src/repository/phase_template_repository.py

@@ -0,0 +1,194 @@
+from sqlalchemy import and_, cast, delete, or_, select, String, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.models.form_type import FormTypeEnum
+from src.db.models.phase_template import PhaseTemplate
+from src.db.models.role import UserRoleEnum
+
+
+_READ_COLUMNS = (
+    PhaseTemplate.form_type.label("form_type"),
+    PhaseTemplate.sheet.label("sheet"),
+    PhaseTemplate.phase_code.label("phase_code"),
+    PhaseTemplate.role.label("role"),
+    PhaseTemplate.column_keys.label("column_keys"),
+    PhaseTemplate.direction.label("direction"),
+    cast(PhaseTemplate.opens_at, String).label("opens_at"),
+    cast(PhaseTemplate.closes_at, String).label("closes_at"),
+)
+
+
+def _row_to_dict(row) -> dict:
+    return dict(row._mapping)
+
+
+class PhaseTemplateRepository:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def get_list(
+            self,
+            form_type: FormTypeEnum | str | None = None,
+            sheet: str | None = None,
+            role: UserRoleEnum | str | None = None,
+            direction: str | None = None,
+    ) -> list[dict]:
+        query = select(*_READ_COLUMNS)
+        if sheet is not None:
+            query = query.where(
+                PhaseTemplate.sheet == sheet,
+                PhaseTemplate.direction.is_(None) if direction is None else PhaseTemplate.direction == direction
+            )
+        if role is not None:
+            query = query.where(PhaseTemplate.role == role)
+        if form_type is not None:
+            query = query.where(PhaseTemplate.form_type == form_type)
+        rows = (await self.db.execute(query)).all()
+        return [_row_to_dict(row) for row in rows]
+
+    async def get(
+        self,
+        form_type: FormTypeEnum | str,
+        sheet: str,
+        direction: str | None,
+        phase_code: str,
+    ) -> dict | None:
+        query = (
+            select(*_READ_COLUMNS)
+            .where(PhaseTemplate.form_type == form_type)
+            .where(PhaseTemplate.sheet == sheet)
+            .where(PhaseTemplate.phase_code == phase_code)
+            .where(PhaseTemplate.direction.is_(None) if direction is None else PhaseTemplate.direction == direction)
+            .limit(1)
+        )
+        row = (await self.db.execute(query)).first()
+        return _row_to_dict(row) if row else None
+
+    async def create(
+        self,
+        form_type: str,
+        sheet: str,
+        phase_code: str,
+        role: UserRoleEnum | str,
+        column_keys: list[str],
+        direction: str | None,
+        opens_at: str,
+        closes_at: str,
+    ) -> dict | None:
+        await self.db.execute(
+            text(
+                """
+                INSERT INTO v3.phase_template
+                    (form_type, sheet, phase_code, role, column_keys, direction, opens_at, closes_at)
+                VALUES
+                    (:form_type, :sheet, :phase_code, :role, :column_keys, :direction,
+                     (:opens_at)::text::timestamp, (:closes_at)::text::timestamp)
+                """
+            ),
+            {
+                "form_type": form_type,
+                "sheet": sheet,
+                "phase_code": phase_code,
+                "role": role,
+                "column_keys": column_keys,
+                "direction": direction,
+                "opens_at": opens_at,
+                "closes_at": closes_at,
+            },
+        )
+        return await self.get(
+            form_type=form_type,
+            sheet=sheet,
+            direction=direction,
+            phase_code=phase_code,
+        )
+
+    async def update(
+        self,
+        form_type: str,
+        sheet: str,
+        phase_code: str,
+        direction: str | None,
+        data: dict,
+    ) -> dict | None:
+        if not data:
+            return await self.get(
+                form_type=form_type,
+                sheet=sheet,
+                direction=direction,
+                phase_code=phase_code,
+            )
+
+        set_parts: list[str] = []
+        params: dict = {
+            "form_type": form_type,
+            "sheet": sheet,
+            "phase_code": phase_code,
+        }
+        if "role" in data:
+            set_parts.append("role = :role")
+            params["role"] = data["role"]
+        if "column_keys" in data:
+            set_parts.append("column_keys = :column_keys")
+            params["column_keys"] = data["column_keys"]
+        if "opens_at" in data:
+            set_parts.append("opens_at = (:opens_at)::text::timestamp")
+            params["opens_at"] = data["opens_at"]
+        if "closes_at" in data:
+            set_parts.append("closes_at = (:closes_at)::text::timestamp")
+            params["closes_at"] = data["closes_at"]
+
+        if direction is None:
+            direction_clause = "direction IS NULL"
+        else:
+            direction_clause = "direction = :direction"
+            params["direction"] = direction
+
+        query = text(
+            f"UPDATE v3.phase_template SET {', '.join(set_parts)} "
+            f"WHERE form_type = :form_type AND sheet = :sheet AND phase_code = :phase_code "
+            f"AND {direction_clause}"
+        )
+        res = await self.db.execute(query, params)
+        if not res.rowcount:
+            return None
+        return await self.get(
+            form_type=form_type,
+            sheet=sheet,
+            direction=direction,
+            phase_code=phase_code,
+        )
+
+    async def delete(
+        self,
+        form_type: FormTypeEnum | str,
+        sheet: str,
+        direction: str | None,
+        phase_code: str,
+    ) -> bool:
+        query = (
+            delete(PhaseTemplate)
+            .where(PhaseTemplate.form_type == form_type)
+            .where(PhaseTemplate.sheet == sheet)
+            .where(PhaseTemplate.phase_code == phase_code)
+            .where(PhaseTemplate.direction.is_(None) if direction is None else PhaseTemplate.direction == direction)
+        )
+        result = await self.db.execute(query)
+        return result.rowcount > 0
+
+    async def bulk_delete(self, lst: list[dict]) -> bool:
+        if not lst:
+            return True
+        clause = or_(
+            *(
+                and_(
+                    PhaseTemplate.form_type == el["form_type"],
+                    PhaseTemplate.sheet == el["sheet"],
+                    PhaseTemplate.phase_code == el["phase_code"],
+                    PhaseTemplate.direction.is_(None) if el["direction"] is None else PhaseTemplate.direction == el["direction"],
+                ) for el in lst
+            )
+        )
+
+        result = await self.db.execute(delete(PhaseTemplate).where(clause))
+        return result.rowcount > 0

api/src/services/phase_template_service.py

@@ -0,0 +1,96 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.repository.phase_template_repository import PhaseTemplateRepository
+from src.db.models.form_type import FormTypeEnum
+from src.db.models.role import UserRoleEnum
+
+
+class PhaseTemplateService:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+        self.pt_repo = PhaseTemplateRepository(db)
+
+    async def get_list(
+            self,
+            form_type: FormTypeEnum | str | None = None,
+            sheet: str | None = None,
+            role: UserRoleEnum | str | None = None,
+            direction: str | None = None,
+    ) -> list[dict]:
+        return await self.pt_repo.get_list(
+            form_type=form_type,
+            sheet=sheet,
+            role=role,
+            direction=direction,
+        )
+
+    async def get(
+        self, 
+        form_type: FormTypeEnum | str,
+        sheet: str,
+        direction: str | None,
+        phase_code: str,
+    ) -> dict | None:
+        return await self.pt_repo.get(
+            form_type=form_type,
+            sheet=sheet,
+            phase_code=phase_code,
+            direction=direction,
+        )
+
+    async def create(
+        self,
+        form_type: str,
+        sheet: str,
+        phase_code: str,
+        role: UserRoleEnum | str,
+        column_keys: list[str],
+        direction: str | None,
+        opens_at: str,
+        closes_at: str,
+    ) -> dict | None:
+        return await self.pt_repo.create(
+            form_type=form_type,
+            sheet=sheet,
+            phase_code=phase_code,
+            direction=direction,
+            role=role,
+            column_keys=column_keys,
+            opens_at=opens_at,
+            closes_at=closes_at,
+        )
+    
+    async def update(
+        self,
+        form_type: str,
+        sheet: str,
+        phase_code: str,
+        direction: str | None,
+        data: dict,
+    ) -> dict | None:
+        return await self.pt_repo.update(
+            form_type=form_type,
+            sheet=sheet,
+            phase_code=phase_code,
+            direction=direction,
+            data=data,
+        )
+        
+    async def delete(
+        self, 
+        form_type: FormTypeEnum | str,
+        sheet: str,
+        direction: str | None,
+        phase_code: str,
+    ) -> bool:
+        return await self.pt_repo.delete(
+            form_type=form_type,
+            sheet=sheet,
+            phase_code=phase_code,
+            direction=direction,
+        )
+
+    async def bulk_delete(self, lst: list[dict]) -> bool:
+        return await self.pt_repo.bulk_delete(
+            lst=lst,
+        )

api/tests/integration/test_phase_template_api_smoke.py

@@ -0,0 +1,178 @@
+def _auth_headers(tokens: dict) -> dict:
+    return {"Authorization": f"Bearer {tokens['access_token']}"}
+
+
+def _create_body(phase_code: str) -> dict:
+    return {
+        "form_type": "FORM_1",
+        "sheet": "OPER",
+        "phase_code": phase_code,
+        "role": "DFIP",
+        "column_keys": ["plan.q1"],
+        "opens_at": "0001-01-01T00:00:00 BC",
+        "closes_at": "0001-02-01T00:00:00 BC",
+        "direction": None,
+    }
+
+
+def test_phase_templates_list_smoke(client, admin_tokens):
+    response = client.get(
+        "/api/v1/stages-templates/", headers=_auth_headers(admin_tokens)
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert "result" in payload
+    assert "count" in payload
+    assert isinstance(payload["result"], list)
+
+
+def test_phase_templates_list_filter_smoke(client, admin_tokens):
+    response = client.get(
+        "/api/v1/stages-templates/", headers=_auth_headers(admin_tokens)
+    )
+    assert response.status_code == 200
+    total = response.json()["count"]
+
+    response = client.get(
+        "/api/v1/stages-templates/?form_type=FORM_1&sheet=AHR",
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert "count" in payload
+    assert payload["count"] <= total
+
+
+def test_phase_templates_create_forbidden(client, isp_tokens):
+    response = client.post(
+        "/api/v1/stages-templates/",
+        json=_create_body("NEW_PHASE"),
+        headers=_auth_headers(isp_tokens),
+    )
+    assert response.status_code == 403
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_phase_templates_update_forbidden(client, isp_tokens):
+    response = client.patch(
+        "/api/v1/stages-templates/FORM_1/OPER/NEW_PHASE",
+        json={"role": "EXECUTOR_RF"},
+        headers=_auth_headers(isp_tokens),
+    )
+    assert response.status_code == 403
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_phase_templates_delete_forbidden(client, isp_tokens):
+    response = client.delete(
+        "/api/v1/stages-templates/FORM_1/OPER/NEW_PHASE",
+        headers=_auth_headers(isp_tokens),
+    )
+    assert response.status_code == 403
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_phase_templates_create(client, admin_tokens):
+    response = client.post(
+        "/api/v1/stages-templates/",
+        json=_create_body("TPL_TEST_CREATE"),
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert "success" in payload
+    assert payload["success"]
+    result = payload["result"]
+    assert isinstance(result, dict)
+    assert result["form_type"] == "FORM_1"
+    assert result["sheet"] == "OPER"
+    assert result["phase_code"] == "TPL_TEST_CREATE"
+    assert result["role"] == "DFIP"
+    assert result["column_keys"] == ["plan.q1"]
+    assert result["direction"] is None
+    assert result["opens_at"]
+    assert result["closes_at"]
+
+
+def test_phase_templates_update(client, admin_tokens):
+    response = client.post(
+        "/api/v1/stages-templates/",
+        json=_create_body("TPL_TEST_UPDATE"),
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+
+    response = client.patch(
+        "/api/v1/stages-templates/FORM_1/OPER/TPL_TEST_UPDATE",
+        json={"role": "EXECUTOR_RF", "column_keys": ["plan.q2"]},
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["success"]
+    result = payload["result"]
+    assert result["phase_code"] == "TPL_TEST_UPDATE"
+    assert result["role"] == "EXECUTOR_RF"
+    assert result["column_keys"] == ["plan.q2"]
+
+
+def test_phase_templates_update_not_found(client, admin_tokens):
+    response = client.patch(
+        "/api/v1/stages-templates/FORM_1/OPER/NON_EXISTENT",
+        json={"role": "EXECUTOR_RF"},
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+
+
+def test_phase_templates_delete(client, admin_tokens):
+    response =  client.post(
+        "/api/v1/stages-templates/",
+        json=_create_body("TPL_TEST_DELETE"),
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+
+    response = client.delete(
+        "/api/v1/stages-templates/FORM_1/OPER/TPL_TEST_DELETE",
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+
+
+def test_phase_templates_delete_not_found(client, admin_tokens):
+    response = client.delete(
+        "/api/v1/stages-templates/FORM_1/OPER/NON_EXISTENT",
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+
+
+def test_phase_templates_bulk_delete(client, admin_tokens):
+    for code in ("TPL_BULK_1", "TPL_BULK_2"):
+        response = client.post(
+            "/api/v1/stages-templates/",
+            json=_create_body(code),
+            headers=_auth_headers(admin_tokens),
+        )
+        assert response.status_code == 200
+
+    response =  client.request(
+        "DELETE",
+        "/api/v1/stages-templates/bulk-delete",
+        json=[
+            {"form_type": "FORM_1", "sheet": "OPER", "phase_code": "TPL_BULK_1", "direction": None},
+            {"form_type": "FORM_1", "sheet": "OPER", "phase_code": "TPL_BULK_2", "direction": None},
+        ],
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["success"]
+    assert payload["result"] is True