RSHB/DFiP_Budget_planing
5
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

vsp-back: перенес функциональность справочников всп

Browse Source
This commit is contained in:
tsygankoviva 2026-06-08 15:06:21 +03:00
parent b6a7ae509e
commit 95d48bb2a0
10 changed files with 1127 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
api/src/api/v1/router.py
View File

@ -1,6 +1,8 @@
from fastapi import APIRouter
from src.api.v1 import auth, users, admin, audit, forms, form_phases, projects, export, org_unit
from src.api.v1 import (
auth, users, admin, audit, forms, form_phases, projects, export, org_unit, vsp,
)
from src.api.v1 import websocket
@ -16,3 +18,4 @@ api_router.include_router(form_phases.router)
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)

183
api/src/api/v1/vsp.py Normal file
View File

@ -0,0 +1,183 @@
from datetime import date
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from src.db.models.app_user import AppUser
from src.api.v1.deps import require_admin, require_executor
from src.db.session import get_db
from src.domain.schemas import (
BaseListResponse,
BaseSingleResponse,
VSPCreate,
VSPExportRequest,
VSPInDB,
VSPUpdate,
)
from src.services.vsp_service import VSPService
router = APIRouter(prefix="/dict/info", tags=["info"])
@router.get("", response_model=BaseListResponse[VSPInDB])
async def get_vsp(
registration_number: str | None = None,
address: str | None = None,
ssp_id: int | None = None,
open_date_start: date | None = None,
open_date_end: date | None = None,
location_form: str | None = None,
numbers_min: int | None = None,
numbers_max: int | None = None,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_executor),
):
"""Получение записей справочника INFO."""
service = VSPService(db)
result = await service.get_list(
user=current_user,
registration_number=registration_number,
address=address,
ssp_id=ssp_id,
open_date_start=open_date_start,
open_date_end=open_date_end,
location_form=location_form,
numbers_min=numbers_min,
numbers_max=numbers_max,
load_org_unit=True,
with_count=True,
)
return BaseListResponse(
success=True,
message="Список INFO",
result=[VSPInDB.model_validate(vsp) for vsp in result[1]],
count=result[0]
)
@router.get("/dropdown", response_model=BaseListResponse[VSPInDB])
async def get_vsp_dropdown(
ssp_id: int | None = None,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_executor),
):
"""Получение записей для выпадающего списка INFO."""
service = VSPService(db)
result = await service.get_dropdown(user=current_user, ssp_id=ssp_id, with_count=True)
return BaseListResponse(
success=True,
message="Список INFO для выпадающего списка",
result=[VSPInDB.model_validate(vsp) for vsp in result[1]],
count=result[0],
)
@router.post("/export")
async def export_info(
body: VSPExportRequest,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_executor),
):
"""Экспорт справочника INFO."""
service = VSPService(db)
stream, filename = await service.export_xlsx(
user=current_user,
registration_number=body.registration_number,
address=body.address,
ssp_ids=body.ssp_ids,
open_date_start=body.open_date_start,
open_date_end=body.open_date_end,
location_form=body.location_form,
numbers_min=body.numbers_min,
numbers_max=body.numbers_max,
)
return StreamingResponse(
stream,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.post("", response_model=BaseSingleResponse[VSPInDB], status_code=201)
async def add_vsp(
body: VSPCreate,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
):
"""Добавление строки в справочник INFO."""
service = VSPService(db)
result = await service.create(
body,
current_user,
load_org_unit=True,
)
return BaseSingleResponse(
success=True,
message="Запись INFO создана",
result=VSPInDB.model_validate(result),
)
@router.get("/{vsp_id}", response_model=BaseSingleResponse[VSPInDB])
async def get_vsp_by_id(
vsp_id: int,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_executor),
):
"""Получение записи справочника INFO по ID."""
service = VSPService(db)
result = await service.get(vsp_id, user=current_user, load_org_unit=True)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"ВСП не найден",
)
return BaseSingleResponse(
success=True,
message="Запись INFO",
result=VSPInDB.model_validate(result),
)
@router.put("/{vsp_id}", response_model=BaseSingleResponse[VSPInDB])
async def update_vsp(
vsp_id: int,
body: VSPUpdate,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_executor),
):
"""Редактирование строки справочника INFO."""
service = VSPService(db)
result = await service.update(
vsp_id,
body,
current_user,
load_org_unit=True,
)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"ВСП не найден",
)
return BaseSingleResponse(
success=True,
message="Запись INFO обновлена",
result=VSPInDB.model_validate(result),
)
@router.delete("/{vsp_id}", response_model=BaseSingleResponse[VSPInDB])
async def delete_vsp(
vsp_id: int,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(require_admin),
):
"""Удаление строки справочника INFO."""
service = VSPService(db)
if await service.logical_delete(vsp_id, current_user):
return BaseSingleResponse(success=True, message="Запись INFO удалена")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"ВСП не найден",
)

2
api/src/db/models/__init__.py
View File

@ -22,7 +22,7 @@ from src.db.models.reserve import Reserve
from src.db.models.rf_project_report import RfProjectReport
from src.db.models.rf_project_report_line import RfProjectReportLine
from src.db.models.rf_project_report_quarter import RfProjectReportQuarter
from src.db.models.role import Role
from src.db.models.role import Role, UserRoleEnum
from src.db.models.security_detail import SecurityDetail
from src.db.models.sequestration import Sequestration
from src.db.models.user_org import UserOrg

2
api/src/db/models/org_unit.py
View File

@ -36,6 +36,6 @@ class OrgUnit(Base):
# projects: Mapped[list["Project"]] = relationship( # type: ignore
# "Project", back_populates="org_unit"
# )
# vsps: Mapped[list["Vsp"]] = relationship("Vsp", back_populates="org_unit") # type: ignore
vsps: Mapped[list["Vsp"]] = relationship("Vsp", back_populates="org_unit") # type: ignore
user_orgs: Mapped[list["UserOrg"]] = relationship("UserOrg", back_populates="org_unit")
users: Mapped[list["AppUser"]] = relationship("AppUser", secondary="v3.user_org", back_populates="org_units")

25
api/src/db/models/vsp.py
View File

@ -1,8 +1,8 @@
from __future__ import annotations
from datetime import date
from datetime import date, datetime
from sqlalchemy import Date, ForeignKey, Integer, Numeric, String
from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Integer, Numeric, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.db.base import Base
@ -22,8 +22,27 @@ class Vsp(Base):
staff_count: Mapped[int | None] = mapped_column(Integer)
total_area: Mapped[float | None] = mapped_column(Numeric)
closed_at: Mapped[date | None] = mapped_column(Date)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
is_deleted: Mapped[bool] = mapped_column(Boolean, default=False)
system_code: Mapped[str | None] = mapped_column(String(100), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
created_by: Mapped[int | None] = mapped_column(Integer, ForeignKey("v3.app_user.id"))
updated_by: Mapped[int | None] = mapped_column(
Integer, ForeignKey("v3.app_user.id"), nullable=True,
)
vsp_type: Mapped[str | None] = mapped_column(String(100), nullable=True)
notes: Mapped[str | None] = mapped_column(Text, nullable=True)
location_form: Mapped[str | None] = mapped_column(String, nullable=True)
numbers: Mapped[int | None] = mapped_column(Integer, nullable=True)
rent_contract_num: Mapped[str | None] = mapped_column(String, nullable=True)
rent_end_date : Mapped[date | None]= mapped_column(Date, nullable=True)
# org_unit = relationship("OrgUnit", back_populates="vsps")
org_unit = relationship("OrgUnit", back_populates="vsps")
# budget_lines: Mapped[list["BudgetLine"]] = relationship("BudgetLine", back_populates="vsp") # type: ignore
# rent_details: Mapped[list["RentDetail"]] = relationship("RentDetail", back_populates="vsp")
# security_details: Mapped[list["SecurityDetail"]] = relationship("SecurityDetail", back_populates="vsp")

15
api/src/domain/enums.py
View File

@ -1,8 +1,23 @@
import enum
#ToDo: вероятно это теперь AuditEvent. А еще он нигде не юзается больше, вероятно, выпилить
class AuditEventType(str, enum.Enum):
LOGIN = "LOGIN"
USER_CREATE = "USER_CREATE"
USER_UPDATE = "USER_UPDATE"
USER_DELETE = "USER_DELETE"
ROW_CREATE = "ROW_CREATE"
ROW_DELETE = "ROW_DELETE"
ACCESS_WINDOW_CHANGE = "ACCESS_WINDOW_CHANGE"
ACCESS_EXTEND = "ACCESS_EXTEND"
USER_ROLE_CHANGE = "USER_ROLE_CHANGE"
CREATE_TASK = "CREATE_TASK"
UPDATE_TASK = "UPDATE_TASK"
DELETE_TASK = "DELETE_TASK"
VSP_CREATE = "VSP_CREATE"
VSP_UPDATE = "VSP_UPDATE"
PROJECT_CREATE = "PROJECT_CREATE"
PROJECT_DELETE = "PROJECT_DELETE"
USER_ACCESS_GRANTED = "USER_ACCESS_GRANTED"
USER_ACCESS_REVOKED = "USER_ACCESS_REVOKED"

158
api/src/domain/schemas.py
View File

@ -1,4 +1,4 @@
from datetime import datetime
from datetime import datetime, date
import enum
from typing import Any, Dict, Generic, List, Literal, Optional, TypeVar
from sqlalchemy.exc import MissingGreenlet
@ -451,3 +451,159 @@ class AuditLogQueryParams(BaseModel):
event_type: Optional[str] = Field(None, description="Тип события")
date_from: Optional[datetime] = Field(None, description="Дата начала (ISO 8601)")
date_to: Optional[datetime] = Field(None, description="Дата окончания (ISO 8601)")
class VSPBase(BaseModel):
system_code: Optional[str] = Field(None, description="Системный код ВСП")
vsp_type: Optional[str] = Field(None, description="Вид ВСП")
registration_number: str = Field(
...,
description="Значение для выпадающего списка",
validation_alias='reg_number',
serialization_alias='registration_number',
)
address: str = Field(..., description="Автоподставляемый адрес")
open_date: Optional[date] = Field(
None,
description="Дата открытия",
validation_alias='opened_at',
serialization_alias='open_date',
)
close_date: Optional[date] = Field(
None,
description="Дата закрытия",
validation_alias='closed_at',
serialization_alias='close_date',
)
approved_format: Optional[str] = Field(
None,
description="Утвержденный формат",
validation_alias='format',
serialization_alias='approved_format',
)
notes: Optional[str] = Field(None, description="Примечания")
location_form: Optional[str] = Field(None, description="Форма расположения")
numbers: Optional[int] = Field(None, description="Штатная численность")
area: Optional[float] = Field(
None,
description="Арендная площадь",
validation_alias='total_area',
serialization_alias='area',
)
rent_contract_num: Optional[str] = Field(None, description="Номер договора аренды")
rent_end_date: Optional[date] = Field(
None, description="Срок окончания договора аренды"
)
ssp_id: int = Field(
...,
description="ID ССП",
validation_alias="branch_id",
serialization_alias="ssp_id",
)
class VSPEditBase(BaseModel):
system_code: Optional[str] = Field(None, description="Системный код ВСП")
vsp_type: Optional[str] = Field(None, description="Вид ВСП")
reg_number: str = Field(
...,
description="Значение для выпадающего списка",
validation_alias='registration_number',
serialization_alias='reg_number',
)
address: str = Field(..., description="Автоподставляемый адрес")
opened_at: Optional[date] = Field(
None,
description="Дата открытия",
validation_alias='open_date',
serialization_alias='opened_at',
)
closed_at: Optional[date] = Field(
None,
description="Дата закрытия",
validation_alias='close_date',
serialization_alias='closed_at',
)
format: Optional[str] = Field(
None,
description="Утвержденный формат",
validation_alias='approved_format',
serialization_alias='format',
)
notes: Optional[str] = Field(None, description="Примечания")
location_form: Optional[str] = Field(None, description="Форма расположения")
numbers: Optional[int] = Field(None, description="Штатная численность")
total_area: Optional[float] = Field(
None,
description="Арендная площадь",
validation_alias='area',
serialization_alias='total_area',
)
rent_contract_num: Optional[str] = Field(None, description="Номер договора аренды")
rent_end_date: Optional[date] = Field(
None, description="Срок окончания договора аренды"
)
branch_id: int = Field(
...,
description="ID ССП",
validation_alias="ssp_id",
serialization_alias="branch_id",
)
class VSPCreate(VSPEditBase):
pass
class VSPUpdate(VSPEditBase):
reg_number: str = Field(
None,
description="Значение для выпадающего списка",
validation_alias='registration_number',
serialization_alias='reg_number',
)
address: str = Field(None, description="Автоподставляемый адрес")
branch_id: int = Field(
None,
description="ID ССП",
validation_alias="ssp_id",
serialization_alias="branch_id",
)
class VSPInDB(VSPBase):
model_config = ConfigDict(from_attributes=True)
id: int
is_active: bool
is_deleted: bool = False
created_by: int
updated_by: Optional[int] = None
created_at: datetime
updated_at: Optional[datetime] = None
regional_branch: Optional[str] = Field(None, description="Региональный филиал")
@model_validator(mode='before')
@classmethod
def get_regional_branch(cls, data: Any) -> Any:
if isinstance(data, dict):
return data
result = vars(data)
result["regional_branch"] = None
try:
result["regional_branch"] = data.org_unit.title
except MissingGreenlet:
pass
return result
class VSPExportRequest(BaseModel):
registration_number: Optional[str] = None
address: Optional[str] = None
ssp_ids: Optional[List[int]] = None
open_date_start: Optional[date] = None
open_date_end: Optional[date] = None
location_form: Optional[str] = None
numbers_min: Optional[int] = None
numbers_max: Optional[int] = None

243
api/src/repository/vsp_repository.py Normal file
View File

@ -0,0 +1,243 @@
from datetime import date
from typing import Optional
from sqlalchemy import func, select, text, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload
from src.db.models import Vsp
from src.domain.schemas import VSPCreate
class VSPRepository:
def __init__(self, db: AsyncSession):
self.db = db
async def get(self, vsp_id: int, is_active: bool | None = None, load_org_unit: bool = False) -> Optional[Vsp]:
queries = [Vsp.id == vsp_id, Vsp.is_deleted.is_(False)]
if is_active is not None:
queries.append(Vsp.is_active == is_active)
if load_org_unit:
return (
(await self.db.execute(
select(Vsp).where(*queries).options(joinedload(Vsp.org_unit)).limit(1))
).scalars().first()
)
else:
return (
(await self.db.execute(select(Vsp).where(*queries).limit(1))).scalars().first()
)
async def get_list(
self,
registration_number: str | None = None,
address: str | None = None,
ssp_ids: list[int] | None = None,
open_date_start: date | None = None,
open_date_end: date | None = None,
location_form: str | None = None,
numbers_min: int | None = None,
numbers_max: int | None = None,
is_active: bool | None = True,
close_date_is_null: bool | None = None,
load_org_unit: bool = False,
with_count: bool = False,
) -> list[Vsp] | tuple[int, list[Vsp]]:
if with_count:
query = select(func.count().over().label("total_count"), Vsp)
else:
query = select(Vsp)
queries = [Vsp.is_deleted.is_(False)]
filters = [
("registration_number", registration_number),
("address", address),
("location_form", location_form),
("open_date_start", open_date_start),
("open_date_end", open_date_end),
("numbers_min", numbers_min),
("numbers_max", numbers_max),
("is_active", is_active),
]
for filter_name, filter_value in filters:
# print(filter_name, filter_value)
match filter_name, filter_value:
case ("registration_number", str(value)) if value:
queries.append(Vsp.reg_number.ilike(f"%{value}%"))
case ("address", str(value)) if value:
queries.append(Vsp.address.ilike(f"%{value}%"))
case ("location_form", str(value)) if value:
queries.append(Vsp.location_form.ilike(f"%{value}%"))
case ("open_date_start", value) if value is not None:
queries.append(Vsp.opened_at >= value)
case ("open_date_end", value) if value is not None:
queries.append(Vsp.opened_at <= value)
case ("numbers_min", value) if value is not None:
queries.append(Vsp.numbers >= value)
case ("numbers_max", value) if value is not None:
queries.append(Vsp.numbers <= value)
case ("is_active", value) if value is not None:
queries.append(Vsp.is_active == value)
case _:
continue
if close_date_is_null is True:
queries.append(Vsp.closed_at.is_(None))
elif close_date_is_null is False:
queries.append(Vsp.closed_at.is_not(None))
if ssp_ids is not None:
if not ssp_ids:
if with_count:
return 0, []
return []
queries.append(Vsp.branch_id.in_(ssp_ids))
query = query.where(*queries)
if load_org_unit:
query = query.options(joinedload(Vsp.org_unit))
query = query.order_by(Vsp.id)
if with_count:
result = (await self.db.execute(query)).all()
if result:
return result[0][0], [res[1] for res in result]
else:
if query.whereclause is not None:
count_query = select(func.count(Vsp.id)).where(
query.whereclause
)
return (await self.db.execute(count_query)).scalar(), []
else:
return (await self.db.execute(query)).scalars().all()
return (await self.db.execute(query.order_by(Vsp.id))).scalars().all()
async def create(self, payload: VSPCreate, created_by: int, load_org_unit: bool = False) -> Vsp:
data = payload.model_dump()
vsp_id = (
await self.db.execute(
text("""
SELECT v3.add_vsp(
:p_branch_id, :p_reg_number, :p_address, :p_format,
:p_opened_at, :p_placement_type, :p_staff_count,
:p_total_area, :p_closed_at, :p_is_active, :p_is_deleted,
:p_system_code, :p_created_by, :p_vsp_type, :p_notes,
:p_location_form, :p_numbers, :p_rent_contract_num,
:p_rent_end_date
)
"""),
{
"p_branch_id": data.get("branch_id"),
"p_reg_number": data.get("reg_number"),
"p_address": data.get("address"),
"p_format": data.get("format"),
"p_opened_at": data.get("opened_at"),
"p_placement_type": data.get("placement_type"),
"p_staff_count": data.get("staff_count"),
"p_total_area": data.get("total_area"),
"p_closed_at": data.get("closed_at"),
"p_is_active": data.get("is_active", True),
"p_is_deleted": data.get("is_deleted", False),
"p_system_code": data.get("system_code"),
"p_created_by": created_by,
"p_vsp_type": data.get("vsp_type"),
"p_notes": data.get("notes"),
"p_location_form": data.get("location_form"),
"p_numbers": data.get("numbers"),
"p_rent_contract_num": data.get("rent_contract_num"),
"p_rent_end_date": data.get("rent_end_date"),
}
)
).scalar()
await self.db.flush()
if load_org_unit:
return (
await self.db.execute(
select(Vsp).where(Vsp.id == vsp_id).options(joinedload(Vsp.org_unit)).limit(1)
)
).scalars().first()
else:
return (
await self.db.execute(
select(Vsp).where(Vsp.id == vsp_id).limit(1)
)
).scalars().first()
async def update(self, vsp: Vsp, data: dict, updated_by: int, load_org_unit: bool = False) -> Vsp:
FIELD_TO_PARAM = {
"system_code": "p_system_code",
"vsp_type": "p_vsp_type",
"reg_number": "p_reg_number",
"address": "p_address",
"opened_at": "p_opened_at",
"closed_at": "p_closed_at",
"format": "p_format",
"notes": "p_notes",
"location_form": "p_location_form",
"numbers": "p_numbers",
"total_area": "p_total_area",
"rent_contract_num": "p_rent_contract_num",
"rent_end_date": "p_rent_end_date",
"branch_id": "p_branch_id",
"is_active": "p_is_active",
"is_deleted": "p_is_deleted",
}
params = {"p_vsp_id": vsp.id, "p_updated_by": updated_by}
for field, param in FIELD_TO_PARAM.items():
params[param] = data.get(field)
params.setdefault("p_placement_type", data.get("placement_type"))
params.setdefault("p_staff_count", data.get("staff_count"))
await self.db.execute(
text("""
SELECT v3.upd_vsp(
:p_vsp_id, :p_branch_id, :p_reg_number, :p_address,
:p_format, :p_opened_at, :p_placement_type, :p_staff_count,
:p_total_area, :p_closed_at, :p_is_active, :p_is_deleted,
:p_system_code, :p_updated_by, :p_vsp_type, :p_notes,
:p_location_form, :p_numbers, :p_rent_contract_num,
:p_rent_end_date
)
"""),
params,
)
await self.db.flush()
await self.db.refresh(vsp)
if load_org_unit:
return (
await self.db.execute(
select(Vsp).where(Vsp.id == vsp.id).options(joinedload(Vsp.org_unit)).limit(1)
)
).scalars().first()
else:
return vsp
async def deactivate_by_close_date(self, as_of: date) -> int:
result = await self.db.execute(
update(Vsp)
.where(
Vsp.is_active.is_(True),
Vsp.closed_at.is_not(None),
Vsp.closed_at <= as_of,
)
.values(is_active=False, updated_at=func.now())
)
await self.db.commit()
return result.rowcount or 0
async def logical_delete(self, vsp_id: int, updated_by: int) -> bool:
result = (
await self.db.execute(
text("SELECT v3.del_vsp(:p_vsp_id, :p_deleted_by)"),
{"p_vsp_id": vsp_id, "p_deleted_by": updated_by},
)
).scalar()
await self.db.flush()
return result is not None

273
api/src/services/vsp_service.py Normal file
View File

@ -0,0 +1,273 @@
import asyncio
import io
from datetime import date, datetime
from typing import Optional
from openpyxl import Workbook
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.errors import AccessDeniedException, ValidationException
from src.db.models import AppUser, UserRoleEnum, Vsp
from src.domain.schemas import VSPCreate, VSPUpdate
from src.repository.user_repository import UserRepository
from src.repository.vsp_repository import VSPRepository
_last_info_close_sync_run: Optional[date] = None
_info_close_sync_lock = asyncio.Lock()
class VSPService:
EXECUTOR_EDITABLE_FIELDS = {
"location_form",
"numbers",
"area",
"rent_contract_num",
"rent_end_date",
}
def __init__(self, db: AsyncSession):
self.db = db
self.vsp_repo = VSPRepository(db)
self.user_repo = UserRepository(db)
async def get(self, vsp_id: int, user: AppUser, load_org_unit: bool = False) -> Optional[Vsp]:
vsp = await self.vsp_repo.get(vsp_id=vsp_id, load_org_unit=load_org_unit)
if not vsp:
return None
if user.role_id == UserRoleEnum.ADMIN:
return vsp
ssp_ids = await self._get_scoped_ssp_ids(user=user)
if not ssp_ids or vsp.ssp_id not in set(ssp_ids):
raise AccessDeniedException()
return vsp
async def get_list(
self,
user: AppUser,
registration_number: str | None = None,
address: str | None = None,
ssp_id: int | None = None,
ssp_ids: list[int] | None = None,
open_date_start: date | None = None,
open_date_end: date | None = None,
location_form: str | None = None,
numbers_min: int | None = None,
numbers_max: int | None = None,
is_active: bool | None = None,
load_org_unit: bool = False,
with_count: bool = False,
) -> list[Vsp] | tuple[int, list[Vsp]]:
await self._sync_closed_records()
scoped_ssp_ids = await self._get_scoped_ssp_ids(user=user)
requested_ssp_ids = list(dict.fromkeys(ssp_ids or []))
if ssp_id is not None and ssp_id not in requested_ssp_ids:
requested_ssp_ids.append(ssp_id)
effective_ssp_ids = scoped_ssp_ids
if requested_ssp_ids:
if scoped_ssp_ids is None:
effective_ssp_ids = requested_ssp_ids
else:
scoped_set = set(scoped_ssp_ids)
effective_ssp_ids = [item for item in requested_ssp_ids if item in scoped_set]
if not effective_ssp_ids:
if with_count:
return 0, []
return []
return await self.vsp_repo.get_list(
registration_number=registration_number,
address=address,
ssp_ids=effective_ssp_ids,
open_date_start=open_date_start,
open_date_end=open_date_end,
location_form=location_form,
numbers_min=numbers_min,
numbers_max=numbers_max,
is_active=is_active,
load_org_unit=load_org_unit,
with_count=with_count,
)
async def get_dropdown(
self,
user: AppUser,
ssp_id: int | None = None,
with_count: bool = False,
) -> list[Vsp]:
ssp_ids = await self._get_scoped_ssp_ids(user=user)
if ssp_id is not None:
if ssp_ids is None:
ssp_ids = [ssp_id]
elif ssp_id not in set(ssp_ids):
if with_count:
return 0, []
return []
else:
ssp_ids = [ssp_id]
return await self.vsp_repo.get_list(
ssp_ids=ssp_ids,
is_active=True,
close_date_is_null=True,
with_count=with_count,
)
async def create(self, payload: VSPCreate, user: AppUser, load_org_unit: bool = False) -> Vsp:
if not self._can_modify(user):
raise AccessDeniedException()
try:
created = await self.vsp_repo.create(payload=payload, created_by=user.id, load_org_unit=load_org_unit)
if created.closed_at is not None and created.closed_at <= date.today():
created = await self.vsp_repo.update(
vsp=created,
data={"is_active": False},
updated_by=user.id,
load_org_unit=load_org_unit,
)
return created
except IntegrityError as exc:
raise ValidationException(description="Некорректные данные для записи ВСП") from exc
async def export_xlsx(
self,
user: AppUser,
registration_number: str | None = None,
address: str | None = None,
ssp_ids: list[int] | None = None,
open_date_start: date | None = None,
open_date_end: date | None = None,
location_form: str | None = None,
numbers_min: int | None = None,
numbers_max: int | None = None,
) -> tuple[io.BytesIO, str]:
data = await self.get_list(
user=user,
registration_number=registration_number,
address=address,
ssp_ids=ssp_ids,
open_date_start=open_date_start,
open_date_end=open_date_end,
location_form=location_form,
numbers_min=numbers_min,
numbers_max=numbers_max,
is_active=None,
load_org_unit=True,
)
workbook = Workbook()
sheet = workbook.active
sheet.title = "INFO"
sheet.append(
[
"id",
"ssp_id",
"system_code",
"regional_branch",
"vsp_type",
"registration_number",
"address",
"approved_format",
"open_date",
"close_date",
"notes",
"location_form",
"numbers",
"area",
"rent_contract_num",
"rent_end_date",
"is_active",
"is_deleted",
"created_at",
"updated_at",
"created_by",
"updated_by",
]
)
for item in data:
sheet.append(
[
item.id,
item.branch_id,
item.system_code or None,
item.org_unit.title or None,
item.vsp_type or None,
item.reg_number,
item.address,
item.format or None,
item.opened_at.isoformat() if item.opened_at else None,
item.closed_at.isoformat() if item.closed_at else None,
item.notes or None,
item.location_form or None,
item.numbers if item.numbers is not None else None,
float(item.total_area) if item.total_area is not None else None,
item.rent_contract_num or None,
item.rent_end_date.isoformat() if item.rent_end_date else None,
item.is_active,
item.is_deleted,
item.created_at.isoformat() if item.created_at else None,
item.updated_at.isoformat() if item.updated_at else None,
item.created_by,
item.updated_by if item.updated_by is not None else None,
]
)
stream = io.BytesIO()
workbook.save(stream)
workbook.close()
stream.seek(0)
file_name = f"info_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return stream, file_name
async def update(self, vsp_id: int, payload: VSPUpdate, user: AppUser, load_org_unit: bool = False) -> Vsp | None:
vsp = await self.vsp_repo.get(vsp_id=vsp_id, is_active=None)
if not vsp:
return None
data = payload.model_dump(exclude_unset=True)
if not data:
return vsp
if user.role_id != UserRoleEnum.ADMIN:
ssp_ids = await self._get_scoped_ssp_ids(user=user)
if not ssp_ids or vsp.ssp_id not in set(ssp_ids):
raise AccessDeniedException()
if not set(data.keys()).issubset(self.EXECUTOR_EDITABLE_FIELDS):
raise AccessDeniedException()
effective_close_date = data.get("close_date", vsp.closed_at)
if effective_close_date is not None and effective_close_date <= date.today():
data["is_active"] = False
try:
return await self.vsp_repo.update(
vsp=vsp, data=data, updated_by=user.id, load_org_unit=load_org_unit,
)
except IntegrityError as exc:
raise ValidationException(description="Некорректные данные для записи ВСП") from exc
async def logical_delete(self, vsp_id: int, user: AppUser) -> bool:
if not self._can_modify(user):
raise AccessDeniedException()
return await self.vsp_repo.logical_delete(vsp_id=vsp_id, updated_by=user.id)
def _can_modify(self, user: AppUser) -> bool:
return user.role_id == UserRoleEnum.ADMIN
async def _get_scoped_ssp_ids(self, user: AppUser) -> list[int] | None:
if user.role_id == UserRoleEnum.ADMIN:
return None
return list(set(await self.user_repo.get_many_ssp_ids(user_id=user.id)))
async def _sync_closed_records(self) -> None:
global _last_info_close_sync_run
today = date.today()
if _last_info_close_sync_run is not None and _last_info_close_sync_run >= today:
return
async with _info_close_sync_lock:
if _last_info_close_sync_run is not None and _last_info_close_sync_run >= today:
return
await self.vsp_repo.deactivate_by_close_date(as_of=today)
_last_info_close_sync_run = today

228
api/tests/integration/test_vsp.py Normal file
View File

@ -0,0 +1,228 @@
import uuid
import pytest
def _unique_reg_number() -> str:
return f"TEST_{uuid.uuid4().hex[:10]}"
def test_vsp_list_smoke(client, admin_tokens, auth_headers):
response = client.get("/api/v1/dict/info", 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_vsp_list_with_filters(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/dict/info?ssp_id=1",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert isinstance(payload["result"], list)
def test_vsp_dropdown_smoke(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/dict/info/dropdown", 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_vsp_dropdown_with_ssp_filter(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/dict/info/dropdown?ssp_id=1",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert isinstance(payload["result"], list)
def test_vsp_get_by_id(client, admin_tokens, auth_headers):
response = client.get("/api/v1/dict/info/1", headers=auth_headers(admin_tokens))
assert response.status_code == 200
payload = response.json()
assert "result" in payload
assert isinstance(payload["result"], dict)
assert payload["result"]["id"] == 1
def test_vsp_get_not_found(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/dict/info/99999", headers=auth_headers(admin_tokens)
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_create_smoke(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
response = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "Test Address",
"ssp_id": 1,
},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 201
payload = response.json()
assert payload["success"] is True
assert payload["result"]["registration_number"] == reg_number
assert payload["result"]["address"] == "Test Address"
assert payload["result"]["ssp_id"] == 1
def test_vsp_create_with_all_fields(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
response = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "Full Address",
"ssp_id": 1,
"system_code": "SYS001",
"vsp_type": "Тип 1",
"open_date": "2024-01-15",
"close_date": "2025-12-31",
"approved_format": "Стандарт",
"notes": "Тестовое примечание",
"location_form": "Встроенное",
"numbers": 10,
"area": 150.5,
"rent_contract_num": "Д-123/24",
"rent_end_date": "2025-12-31",
},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 201
payload = response.json()
assert payload["success"] is True
assert payload["result"]["registration_number"] == reg_number
assert payload["result"]["system_code"] == "SYS001"
assert payload["result"]["vsp_type"] == "Тип 1"
assert payload["result"]["open_date"] == "2024-01-15"
assert payload["result"]["close_date"] == "2025-12-31"
assert payload["result"]["approved_format"] == "Стандарт"
assert payload["result"]["location_form"] == "Встроенное"
assert payload["result"]["numbers"] == 10
assert payload["result"]["area"] == 150.5
assert payload["result"]["rent_contract_num"] == "Д-123/24"
assert payload["result"]["rent_end_date"] == "2025-12-31"
def test_vsp_create_forbidden_for_executor(client, isp_tokens, auth_headers):
response = client.post(
"/api/v1/dict/info",
json={
"registration_number": _unique_reg_number(),
"address": "Should Fail",
"ssp_id": 1,
},
headers=auth_headers(isp_tokens),
)
assert response.status_code == 403
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_update_smoke(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
created = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "Original Address",
"ssp_id": 1,
},
headers=auth_headers(admin_tokens),
)
assert created.status_code == 201
vsp_id = created.json()["result"]["id"]
new_address = f"Updated Address {uuid.uuid4().hex[:4]}"
response = client.put(
f"/api/v1/dict/info/{vsp_id}",
json={"address": new_address},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert payload["success"] is True
assert payload["result"]["address"] == new_address
def test_vsp_update_not_found(client, admin_tokens, auth_headers):
response = client.put(
"/api/v1/dict/info/99999",
json={"address": "No Matter"},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_delete_smoke(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
created = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "To Be Deleted",
"ssp_id": 1,
},
headers=auth_headers(admin_tokens),
)
assert created.status_code == 201
vsp_id = created.json()["result"]["id"]
response = client.delete(
f"/api/v1/dict/info/{vsp_id}",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert payload["success"] is True
assert payload["message"] == "Запись INFO удалена"
def test_vsp_delete_not_found(client, admin_tokens, auth_headers):
response = client.delete(
"/api/v1/dict/info/99999",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_delete_forbidden_for_executor(client, isp_tokens, auth_headers):
response = client.delete(
"/api/v1/dict/info/1",
headers=auth_headers(isp_tokens),
)
assert response.status_code == 403
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_requires_auth(client):
response = client.get("/api/v1/dict/info")
assert response.status_code == 403