a7da43be0e
- session: set_tenant use bound param (SQL injection fix)
- health: text('SELECT 1'), REDIS_URL from config
- deps: re-export get_db from session, use settings.ENABLE_DEV_AUTH (default False)
- routes: all get_db from app.api.deps; conftest overrides deps.get_db
- main: register exception handlers from app.api.exceptions
- next.config: enable ESLint and TypeScript checks
- .eslintrc: drop @typescript-eslint/recommended; fix no-console (logger, ws-client, regulations)
- backend/.env.example added
- frontend: export apiFetch; dashboard, profile, settings, risks use api-client
- docs/ANALYSIS_AND_RECOMMENDATIONS.md
Co-authored-by: Cursor <cursoragent@cursor.com>
106 lines
6.0 KiB
Python
106 lines
6.0 KiB
Python
"""Checklists API — refactored: pagination, audit, DRY."""
|
||
from fastapi import APIRouter, Depends, HTTPException, File, UploadFile, Query
|
||
from sqlalchemy.orm import Session
|
||
import csv, io
|
||
|
||
from app.api.deps import get_current_user, require_roles
|
||
from app.api.helpers import audit, paginate_query
|
||
from app.api.deps import get_db
|
||
from app.models import ChecklistTemplate, ChecklistItem
|
||
from app.schemas.audit import ChecklistTemplateCreate, ChecklistTemplateOut, ChecklistItemCreate, ChecklistItemOut
|
||
|
||
router = APIRouter(tags=["checklists"])
|
||
|
||
|
||
def _template_with_items(template, db) -> ChecklistTemplateOut:
|
||
items = db.query(ChecklistItem).filter(ChecklistItem.template_id == template.id).order_by(ChecklistItem.sort_order).all()
|
||
out = ChecklistTemplateOut.model_validate(template)
|
||
out.items = [ChecklistItemOut.model_validate(i) for i in items]
|
||
return out
|
||
|
||
|
||
@router.get("/checklists/templates")
|
||
def list_templates(
|
||
domain: str | None = None, page: int = Query(1, ge=1), per_page: int = Query(25, ge=1, le=100),
|
||
db: Session = Depends(get_db), user=Depends(get_current_user),
|
||
):
|
||
q = db.query(ChecklistTemplate).filter(ChecklistTemplate.is_active == True)
|
||
if domain: q = q.filter(ChecklistTemplate.domain == domain)
|
||
q = q.order_by(ChecklistTemplate.name, ChecklistTemplate.version.desc())
|
||
result = paginate_query(q, page, per_page)
|
||
result["items"] = [_template_with_items(t, db) for t in result["items"]]
|
||
return result
|
||
|
||
|
||
@router.post("/checklists/templates", response_model=ChecklistTemplateOut, status_code=201,
|
||
dependencies=[Depends(require_roles("admin", "authority_inspector"))])
|
||
def create_template(payload: ChecklistTemplateCreate, db: Session = Depends(get_db), user=Depends(get_current_user)):
|
||
template = ChecklistTemplate(name=payload.name, version=payload.version, description=payload.description, domain=payload.domain)
|
||
db.add(template); db.flush()
|
||
if payload.items:
|
||
for d in payload.items:
|
||
db.add(ChecklistItem(template_id=template.id, code=d.code, text=d.text, domain=d.domain, sort_order=d.sort_order))
|
||
audit(db, user, "create", "checklist_template", description=f"Template: {payload.name}")
|
||
db.commit(); db.refresh(template)
|
||
return _template_with_items(template, db)
|
||
|
||
|
||
@router.post("/checklists/generate", response_model=ChecklistTemplateOut,
|
||
dependencies=[Depends(require_roles("admin", "authority_inspector"))])
|
||
def generate_checklist(source: str, name: str, items: list[ChecklistItemCreate] | None = None,
|
||
db: Session = Depends(get_db), user=Depends(get_current_user)):
|
||
template = ChecklistTemplate(name=name, version=1, domain=source)
|
||
db.add(template); db.flush()
|
||
if source == "fap_m_inspection":
|
||
items = [
|
||
ChecklistItemCreate(code="M.A.301", text="ВС имеет действующий сертификат лётной годности", domain="ФАП-М"),
|
||
ChecklistItemCreate(code="M.A.302", text="ВС эксплуатируется в соответствии с CAME", domain="ФАП-М"),
|
||
ChecklistItemCreate(code="M.A.303", text="Выполнены все требования по ТО", domain="ФАП-М"),
|
||
ChecklistItemCreate(code="M.A.304", text="Все дефекты устранены или имеют разрешения", domain="ФАП-М"),
|
||
ChecklistItemCreate(code="M.A.305", text="Документация по ВС актуальна", domain="ФАП-М"),
|
||
]
|
||
elif source != "custom" or not items:
|
||
raise HTTPException(400, "Invalid source or missing items")
|
||
for d in items:
|
||
db.add(ChecklistItem(template_id=template.id, code=d.code, text=d.text, domain=d.domain, sort_order=d.sort_order))
|
||
audit(db, user, "create", "checklist_template", description=f"Generated: {name}")
|
||
db.commit(); db.refresh(template)
|
||
return _template_with_items(template, db)
|
||
|
||
|
||
@router.get("/checklists/templates/{template_id}", response_model=ChecklistTemplateOut)
|
||
def get_template(template_id: str, db: Session = Depends(get_db), user=Depends(get_current_user)):
|
||
t = db.query(ChecklistTemplate).filter(ChecklistTemplate.id == template_id).first()
|
||
if not t: raise HTTPException(404, "Not found")
|
||
return _template_with_items(t, db)
|
||
|
||
|
||
@router.post("/checklists/generate-from-csv", response_model=ChecklistTemplateOut,
|
||
dependencies=[Depends(require_roles("admin", "authority_inspector"))])
|
||
async def generate_from_csv(file: UploadFile = File(...), name: str | None = None, domain: str | None = None,
|
||
db: Session = Depends(get_db), user=Depends(get_current_user)):
|
||
content = await file.read()
|
||
try: text = content.decode('utf-8-sig')
|
||
except UnicodeDecodeError: text = content.decode('cp1251')
|
||
reader = csv.DictReader(io.StringIO(text))
|
||
fields = reader.fieldnames or []
|
||
rows = list(reader)
|
||
if not rows: raise HTTPException(400, "CSV empty")
|
||
code_col = next((c for c in fields if 'issue' in c.lower() and 'id' in c.lower()), None)
|
||
text_col = next((c for c in fields if 'summary' in c.lower() or 'description' in c.lower()), None)
|
||
if not code_col or not text_col: raise HTTPException(400, "Missing Issue Id / Summary columns")
|
||
domain_col = next((c for c in fields if 'domain' in c.lower()), None)
|
||
template = ChecklistTemplate(name=name or file.filename.replace('.csv', ''), version=1, domain=domain or "CSV")
|
||
db.add(template); db.flush()
|
||
count = 0
|
||
for idx, row in enumerate(rows):
|
||
code, txt = str(row.get(code_col, "")).strip(), str(row.get(text_col, "")).strip()
|
||
if not code or not txt: continue
|
||
db.add(ChecklistItem(template_id=template.id, code=code, text=txt,
|
||
domain=domain or (row.get(domain_col, "") if domain_col else None), sort_order=idx+1))
|
||
count += 1
|
||
if not count: db.rollback(); raise HTTPException(400, "No items created")
|
||
audit(db, user, "create", "checklist_template", description=f"CSV import: {count} items")
|
||
db.commit(); db.refresh(template)
|
||
return _template_with_items(template, db)
|