expense-item-handler: получение expense-item по id

api/src/api/v1/expense_item.py

@@ -0,0 +1,35 @@
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.services.expense_item_service import ExpenseItemService
+from src.db.models.app_user import AppUser
+from src.api.v1.deps import require_admin
+from src.db.session import get_db
+from src.domain.schemas import (
+    BaseSingleResponse,
+    ExpenseItemResponseSchema,
+)
+
+router = APIRouter(prefix="/expense-item", tags=["expense-item"])
+
+
+@router.get("/{item_id}", response_model=BaseSingleResponse[ExpenseItemResponseSchema])
+async def get_expense_item_by_id(
+    item_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+):
+    """Получение записи expense item по ID."""
+    service = ExpenseItemService(db)
+    result = await service.get(item_id, user=current_user)
+    if not result:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Expense Item не найден",
+        )
+    return BaseSingleResponse(
+        success=True,
+        message="Запись Expense Item",
+        result=ExpenseItemResponseSchema.model_validate(result),
+    )

api/src/api/v1/router.py

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

api/src/db/models/__init__.py

@@ -9,6 +9,7 @@ from src.db.models.ckk import Ckk
 from src.db.models.collegial_approval import CollegialApproval
 from src.db.models.contract_detail import ContractDetail
 from src.db.models.contract_summary import ContractSummary
+from src.db.models.expense_item import ExpenseItem
 from src.db.models.form_limit import FormLimit
 from src.db.models.form_phase import FormPhase
 from src.db.models.form_type import FormType

api/src/db/models/expense_item.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from sqlalchemy import CheckConstraint, ForeignKey, Index, Integer, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.db.base import Base
+
+
+class ExpenseItem(Base):
+    __tablename__ = "expense_item"
+    __table_args__ = (
+        CheckConstraint("direction IN ('Support', 'Development')", name="expense_item_direction_check"),
+        CheckConstraint("sheet IN ('AHR', 'CAP', 'OPER')", name="expense_item_sheet_check"),
+        Index("ix_v3_expense_item_parent", "parent_id"),
+        Index("ix_v3_expense_item_sheet", "sheet"),
+        {"schema": "v3"},
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    section_code: Mapped[str | None] = mapped_column(String)
+    item_id: Mapped[str | None] = mapped_column(String)
+    num_group_id: Mapped[str | None] = mapped_column(String)
+    name: Mapped[str | None] = mapped_column(String)
+    sheet: Mapped[str | None] = mapped_column(String)
+    direction: Mapped[str | None] = mapped_column(String)
+    parent_id: Mapped[int | None] = mapped_column(Integer, ForeignKey("v3.expense_item.id"))
+    depth: Mapped[int | None] = mapped_column(Integer, default=0)

api/src/domain/schemas.py

@@ -613,3 +613,21 @@ class VSPExportRequest(BaseModel):
     location_form: Optional[str] = None
     numbers_min: Optional[int] = None
     numbers_max: Optional[int] = None
+
+
+class ExpenseItemBaseSchema(BaseModel):    
+    section_code: Optional[str] = None
+    item_id: Optional[str] = None
+    num_group_id: Optional[str] = None
+    name: Optional[str] = None
+    sheet: Optional[str] = None
+    direction: Optional[str] = None
+    parent_id: Optional[int] = None
+    depth: Optional[int] = Field(default=0)
+
+    model_config = {
+        "from_attributes": True
+    }
+
+class ExpenseItemResponseSchema(ExpenseItemBaseSchema):
+    id: int

api/src/repository/expense_item_repository.py

@@ -0,0 +1,13 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.models.expense_item import ExpenseItem
+
+
+class ExpenseItemRepository:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def get(self, item_id: int) -> ExpenseItem | None:
+        query = select(ExpenseItem).where(ExpenseItem.id == item_id).limit(1)
+        return (await self.db.execute(query)).scalar_one_or_none()

api/src/services/expense_item_service.py

@@ -0,0 +1,16 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.models.role import UserRoleEnum
+from src.db.models.app_user import AppUser
+from src.db.models.expense_item import ExpenseItem
+from src.repository.expense_item_repository import ExpenseItemRepository
+
+
+class ExpenseItemService:
+    def __init__(self, db: AsyncSession):
+        self.repo = ExpenseItemRepository(db)
+
+    async def get(self, item_id: int, user: AppUser | None = None) -> ExpenseItem | None:
+        if not user or user.role_id == UserRoleEnum.ADMIN:
+            return await self.repo.get(item_id=item_id)
+        return None

api/tests/integration/test_expense_item.py

@@ -0,0 +1,32 @@
+import pytest
+
+def test_expense_item_get_found(client, admin_tokens, auth_headers):
+    response = client.get(
+        "/api/v1/expense-item/1",
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    result = response.json()
+    assert result["success"] is True
+    assert result["result"]["id"] == 1
+
+
+def test_expense_item_get_not_found(client, admin_tokens, auth_headers):
+    response = client.get(
+        "/api/v1/expense-item/999999",
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+
+
+def test_expense_item_get_unauthorized(client):
+    response = client.get("/api/v1/expense-item/1")
+    assert response.status_code == 403
+
+
+def test_expense_item_get_executor(client, isp_tokens, auth_headers):
+    response = client.get(
+        "/api/v1/expense-item/1",
+        headers=auth_headers(isp_tokens),
+    )
+    assert response.status_code == 403