97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
"""
|
|
Lightweight form metadata API for AI workflow panels.
|
|
"""
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.base_schema import PaginatedResponse
|
|
from app.database import get_db
|
|
from core.application.model import Application
|
|
from online_dev.form_manager.model import FormMeta
|
|
|
|
router = APIRouter(prefix="/forms", tags=["AI-Forms"])
|
|
|
|
|
|
def _build_form_list_out(form, application_name: str = "", application_code: str = "") -> dict:
|
|
return {
|
|
"id": str(form.id),
|
|
"application_id": form.application_id,
|
|
"application_name": application_name or "Main App",
|
|
"application_code": application_code or "",
|
|
"name": form.name,
|
|
"code": form.code,
|
|
"status": form.status,
|
|
}
|
|
|
|
|
|
@router.get("/list", summary="List form metadata for AI workflow form-data nodes")
|
|
async def list_forms(
|
|
application_id: Optional[str] = Query(None, alias="applicationId"),
|
|
name: Optional[str] = Query(None),
|
|
code: Optional[str] = Query(None),
|
|
status: Optional[str] = Query(None),
|
|
include_globally_visible: bool = Query(False, alias="includeGloballyVisible"),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100, alias="pageSize"),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
conditions = [FormMeta.is_deleted.is_(False)]
|
|
|
|
if application_id:
|
|
app_condition = FormMeta.application_id == application_id
|
|
if include_globally_visible:
|
|
app_condition = or_(
|
|
app_condition,
|
|
and_(
|
|
FormMeta.globally_visible.is_(True),
|
|
FormMeta.status == "published",
|
|
),
|
|
)
|
|
conditions.append(app_condition)
|
|
elif not include_globally_visible:
|
|
conditions.append(FormMeta.application_id.is_(None))
|
|
else:
|
|
conditions.append(
|
|
or_(
|
|
FormMeta.application_id.is_(None),
|
|
and_(
|
|
FormMeta.globally_visible.is_(True),
|
|
FormMeta.status == "published",
|
|
),
|
|
)
|
|
)
|
|
|
|
if name:
|
|
conditions.append(FormMeta.name.ilike(f"%{name}%"))
|
|
if code:
|
|
conditions.append(FormMeta.code.ilike(f"%{code}%"))
|
|
if status:
|
|
conditions.append(FormMeta.status == status)
|
|
|
|
count_stmt = select(func.count(FormMeta.id)).where(and_(*conditions))
|
|
total = (await db.execute(count_stmt)).scalar() or 0
|
|
|
|
offset = (page - 1) * page_size
|
|
stmt = (
|
|
select(
|
|
FormMeta,
|
|
Application.name.label("application_name"),
|
|
Application.code.label("application_code"),
|
|
)
|
|
.outerjoin(Application, FormMeta.application_id == Application.id)
|
|
.where(and_(*conditions))
|
|
.order_by(FormMeta.sort, FormMeta.sys_create_datetime.desc())
|
|
.offset(offset)
|
|
.limit(page_size)
|
|
)
|
|
result = await db.execute(stmt)
|
|
|
|
items = [
|
|
_build_form_list_out(form, application_name, application_code)
|
|
for form, application_name, application_code in result
|
|
]
|
|
return PaginatedResponse(items=items, total=total)
|