fers web

route/init_data_base.py

@@ -7,6 +7,7 @@ from typing import List
 from bd import Settings
 from sqlalchemy import create_engine
 from sqlalchemy.exc import SQLAlchemyError
+from pydantic import BaseModel
 
 router = APIRouter(tags=["db"])
 
@@ -67,3 +68,40 @@ async def create_tables():
         return {"status": "ok", "detail": f"Created {len(unique)} metadata groups"}
     except SQLAlchemyError as e:
         raise HTTPException(status_code=500, detail=str(e))
+
+
+class ClearDBIn(BaseModel):
+    confirm: bool
+
+
+@router.post("/db/clear")
+async def clear_tables(payload: ClearDBIn):
+    """Полная очистка всех таблиц, описанных в `bd.tables`.
+
+    Требуется явное подтверждение: POST с телом {"confirm": true}.
+    """
+    if not payload.confirm:
+        raise HTTPException(status_code=400, detail="Confirmation required")
+
+    settings = Settings()
+    db_url = settings.DATABASE_URL_syncpg
+    engine = create_engine(db_url, future=True)
+
+    metadatas = collect_metadatas()
+    if not metadatas:
+        raise HTTPException(status_code=400, detail="No table metadata found in db.tables")
+
+    # deduplicate metadata objects
+    unique = []
+    seen = set()
+    for md in metadatas:
+        if id(md) not in seen:
+            seen.add(id(md))
+            unique.append(md)
+
+    try:
+        for md in unique:
+            md.drop_all(bind=engine)
+        return {"status": "ok", "detail": f"Dropped {len(unique)} metadata groups"}
+    except SQLAlchemyError as e:
+        raise HTTPException(status_code=500, detail=str(e))