[fix]:[FL-70][删除线路时,塔杆要一起删除]

Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
chengkai3
2026-06-09 12:25:15 +08:00
parent d7f712e3c1
commit 4ce57708b4
6 changed files with 1782 additions and 17 deletions
+2 -8
View File
@@ -88,15 +88,9 @@ def delete_line_endpoint(
_: CurrentUser = Depends(require_permission("line.manage")),
db: Session = Depends(get_db),
) -> dict[str, bool]:
deleted, tower_count = delete_line(db, line_id)
deleted = delete_line(db, line_id)
if not deleted:
line = get_line_by_id(db, line_id)
if not line:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Line not found")
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Line has {tower_count} towers, delete towers first",
)
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Line not found")
return {"success": True}
+9
View File
@@ -12,6 +12,7 @@ from .base import utcnow
if TYPE_CHECKING:
from .line import Line
from .tower_profile import TowerProfile
class LineTower(Base):
@@ -63,3 +64,11 @@ class LineTower(Base):
update_user: Mapped[str | None] = mapped_column(String(64), index=True)
line: Mapped[Line] = relationship("Line", back_populates="towers", lazy="selectin")
profile: Mapped[TowerProfile | None] = relationship(
"TowerProfile",
back_populates="tower",
lazy="selectin",
cascade="all, delete-orphan",
single_parent=True,
uselist=False,
)
+1 -1
View File
@@ -64,4 +64,4 @@ class TowerProfile(Base):
)
update_user: Mapped[str | None] = mapped_column(String(64), index=True)
tower: Mapped[LineTower] = relationship("LineTower", lazy="selectin")
tower: Mapped[LineTower] = relationship("LineTower", back_populates="profile", lazy="selectin")
+3 -7
View File
@@ -225,19 +225,15 @@ def update_line(
)
def delete_line(db: Session, line_id: str) -> tuple[bool, int]:
def delete_line(db: Session, line_id: str) -> bool:
line = get_line_by_id(db, line_id)
if not line:
return False, 0
tower_count = int(db.scalar(select(func.count()).select_from(LineTower).where(LineTower.line_id == line_id)) or 0)
if tower_count > 0:
return False, tower_count
return False
db.delete(line)
db.commit()
_publish_line_change("power-lines.deleted", {"action": "deleted", "line_id": line_id})
return True, 0
return True
def list_line_towers(
+42 -1
View File
@@ -4,7 +4,7 @@ import csv
import io
from types import SimpleNamespace
from sqlalchemy import create_engine
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
from app.core.database import Base
@@ -73,6 +73,47 @@ def test_generate_line_code_skips_existing_code(monkeypatch) -> None:
session.close()
def test_delete_line_cascades_to_towers_and_profiles(monkeypatch) -> None:
session = _build_session()
try:
monkeypatch.setattr(line_service, "_publish_line_change", lambda *args, **kwargs: None)
line = Line(
code="PL-DELETE-001",
name="待删除线路",
status="enabled",
)
session.add(line)
session.flush()
tower = LineTower(
line_id=line.id,
seq_no=1,
tower_no="T-001",
tower_model="ZM-001",
tower_type="直线塔",
)
session.add(tower)
session.flush()
session.add(
TowerProfile(
tower_id=tower.id,
structure_kind="直线塔",
)
)
session.commit()
deleted = line_service.delete_line(session, line.id)
assert deleted is True
assert line_service.get_line_by_id(session, line.id) is None
assert session.execute(select(LineTower).where(LineTower.line_id == line.id)).scalar_one_or_none() is None
assert session.execute(select(TowerProfile).where(TowerProfile.tower_id == tower.id)).scalar_one_or_none() is None
finally:
session.close()
def test_export_line_towers_to_csv_includes_legacy_professional_columns() -> None:
session = _build_session()
try:
Generated
+1725
View File
File diff suppressed because it is too large Load Diff