367 lines
No EOL
15 KiB
Python
367 lines
No EOL
15 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
from fastapi import APIRouter, HTTPException, Depends, Body, Path, Request, Query
|
|
from typing import List, Dict, Any, Optional, Tuple
|
|
from fastapi import status
|
|
from fastapi.responses import JSONResponse
|
|
import logging
|
|
import json
|
|
import math
|
|
from copy import deepcopy
|
|
|
|
# Import auth module
|
|
from modules.auth import limiter, getCurrentUser
|
|
|
|
# Import interfaces
|
|
import modules.interfaces.interfaceDbManagement as interfaceDbManagement
|
|
from modules.datamodels.datamodelUtils import Prompt
|
|
from modules.datamodels.datamodelUam import User
|
|
from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResponse, PaginationMetadata, normalize_pagination_dict
|
|
from modules.shared.i18nRegistry import apiRouteContext
|
|
routeApiMsg = apiRouteContext("routeDataPrompts")
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Create router for prompt endpoints
|
|
router = APIRouter(
|
|
prefix="/api/prompts",
|
|
tags=["Manage Prompts"],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
@router.get("")
|
|
@limiter.limit("30/minute")
|
|
def get_prompts(
|
|
request: Request,
|
|
pagination: Optional[str] = Query(None, description="JSON-encoded PaginationParams object"),
|
|
mode: Optional[str] = Query(None, description="'filterValues' for distinct column values, 'ids' for all filtered IDs"),
|
|
column: Optional[str] = Query(None, description="Column key (required when mode=filterValues)"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
):
|
|
"""
|
|
Get prompts with optional pagination, sorting, and filtering.
|
|
|
|
Modes:
|
|
- None: paginated list (default)
|
|
- filterValues: distinct values for a column (cross-filtered)
|
|
- ids: all IDs matching current filters
|
|
"""
|
|
from modules.routes.routeHelpers import (
|
|
handleFilterValuesInMemory, handleIdsInMemory, enrichRowsWithFkLabels,
|
|
resolveView, applyViewToParams, buildGroupLayout, effective_group_by_levels,
|
|
)
|
|
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
|
|
from modules.datamodels.datamodelPagination import AppliedViewMeta
|
|
|
|
CONTEXT_KEY = "prompts"
|
|
|
|
paginationParams = None
|
|
if pagination:
|
|
try:
|
|
paginationDict = json.loads(pagination)
|
|
if paginationDict:
|
|
paginationDict = normalize_pagination_dict(paginationDict)
|
|
paginationParams = PaginationParams(**paginationDict)
|
|
except (json.JSONDecodeError, ValueError) as e:
|
|
raise HTTPException(status_code=400, detail=f"Invalid pagination parameter: {str(e)}")
|
|
|
|
appInterface = getAppInterface(currentUser)
|
|
|
|
# Resolve view and merge config into params
|
|
viewKey = paginationParams.viewKey if paginationParams else None
|
|
viewConfig, viewDisplayName = resolveView(appInterface, CONTEXT_KEY, viewKey)
|
|
viewMeta = AppliedViewMeta(viewKey=viewKey, displayName=viewDisplayName) if viewKey else None
|
|
paginationParams = applyViewToParams(paginationParams, viewConfig)
|
|
groupByLevels = effective_group_by_levels(paginationParams, viewConfig)
|
|
|
|
def _getVisiblePromptSearchFields() -> List[str]:
|
|
"""Return Prompt fields considered visible/searchable in the table."""
|
|
fields: List[str] = []
|
|
for fieldName, fieldInfo in Prompt.model_fields.items():
|
|
if fieldName == "id":
|
|
continue
|
|
jsonExtra = fieldInfo.json_schema_extra or {}
|
|
if jsonExtra.get("frontend_visible", True) is False:
|
|
continue
|
|
if jsonExtra.get("system", False):
|
|
continue
|
|
fields.append(fieldName)
|
|
return fields
|
|
|
|
visibleSearchFields = _getVisiblePromptSearchFields()
|
|
|
|
def _applyPromptSearchOnVisibleFields(rows: List[Dict[str, Any]], searchValue: Any) -> List[Dict[str, Any]]:
|
|
"""Apply global prompt search only on visible Prompt fields."""
|
|
searchTerm = str(searchValue or "").strip().lower()
|
|
if not searchTerm:
|
|
return rows
|
|
filteredRows: List[Dict[str, Any]] = []
|
|
for row in rows:
|
|
for fieldName in visibleSearchFields:
|
|
value = row.get(fieldName)
|
|
if value is None:
|
|
continue
|
|
if searchTerm in str(value).lower():
|
|
filteredRows.append(row)
|
|
break
|
|
return filteredRows
|
|
|
|
def _extractSearchAndRemoveFromFilters(filters: Optional[Dict[str, Any]]) -> Tuple[Any, Optional[Dict[str, Any]]]:
|
|
"""Extract generic search from filters and return remaining filters."""
|
|
if not filters:
|
|
return None, None
|
|
cleanedFilters = deepcopy(filters)
|
|
searchValue = cleanedFilters.pop("search", None)
|
|
return searchValue, (cleanedFilters if cleanedFilters else None)
|
|
|
|
def _promptsToEnrichedDicts(promptItems):
|
|
dicts = [r.model_dump() if hasattr(r, 'model_dump') else (dict(r) if not isinstance(r, dict) else r) for r in promptItems]
|
|
enrichRowsWithFkLabels(dicts, Prompt)
|
|
return dicts
|
|
|
|
managementInterface = interfaceDbManagement.getInterface(currentUser)
|
|
|
|
if mode == "groupSummary":
|
|
if not pagination:
|
|
raise HTTPException(status_code=400, detail="pagination required for groupSummary")
|
|
from modules.routes.routeHelpers import (
|
|
apply_strategy_b_filters_and_sort,
|
|
build_group_summary_groups,
|
|
)
|
|
if not groupByLevels or not groupByLevels[0].get("field"):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="groupByLevels[0].field required for groupSummary",
|
|
)
|
|
field = groupByLevels[0]["field"]
|
|
null_label = str(groupByLevels[0].get("nullLabel") or "—")
|
|
result = managementInterface.getAllPrompts(pagination=None)
|
|
allItems = _promptsToEnrichedDicts(
|
|
result if isinstance(result, list) else (result.items if hasattr(result, "items") else [])
|
|
)
|
|
filtered = apply_strategy_b_filters_and_sort(allItems, paginationParams, currentUser)
|
|
groups_out = build_group_summary_groups(filtered, field, null_label, groupByLevels=groupByLevels)
|
|
return JSONResponse(content={"groups": groups_out})
|
|
|
|
if mode == "filterValues":
|
|
if not column:
|
|
raise HTTPException(status_code=400, detail="column parameter required for mode=filterValues")
|
|
result = managementInterface.getAllPrompts(pagination=None)
|
|
return handleFilterValuesInMemory(_promptsToEnrichedDicts(result), column, pagination)
|
|
|
|
if mode == "ids":
|
|
result = managementInterface.getAllPrompts(pagination=None)
|
|
return handleIdsInMemory(_promptsToEnrichedDicts(result), pagination)
|
|
|
|
if not groupByLevels:
|
|
# No grouping: let DB handle pagination directly
|
|
promptSearchValue = None
|
|
remainingFilters = paginationParams.filters if paginationParams else None
|
|
if paginationParams and paginationParams.filters:
|
|
promptSearchValue, remainingFilters = _extractSearchAndRemoveFromFilters(paginationParams.filters)
|
|
|
|
# For prompt search, we must filter before page slicing.
|
|
if paginationParams and promptSearchValue is not None:
|
|
result = managementInterface.getAllPrompts(pagination=None)
|
|
allItems = _promptsToEnrichedDicts(result if isinstance(result, list) else (result.items if hasattr(result, 'items') else []))
|
|
allItems = _applyPromptSearchOnVisibleFields(allItems, promptSearchValue)
|
|
|
|
if remainingFilters or paginationParams.sort:
|
|
from modules.interfaces.interfaceDbManagement import ComponentObjects
|
|
comp = ComponentObjects()
|
|
comp.setUserContext(currentUser)
|
|
if remainingFilters:
|
|
allItems = comp._applyFilters(allItems, remainingFilters)
|
|
if paginationParams.sort:
|
|
allItems = comp._applySorting(allItems, paginationParams.sort)
|
|
|
|
totalItems = len(allItems)
|
|
totalPages = math.ceil(totalItems / paginationParams.pageSize) if totalItems > 0 else 0
|
|
startIdx = (paginationParams.page - 1) * paginationParams.pageSize
|
|
endIdx = startIdx + paginationParams.pageSize
|
|
response = {
|
|
"items": allItems[startIdx:endIdx],
|
|
"pagination": PaginationMetadata(
|
|
currentPage=paginationParams.page,
|
|
pageSize=paginationParams.pageSize,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages,
|
|
sort=paginationParams.sort,
|
|
filters=paginationParams.filters
|
|
).model_dump(),
|
|
}
|
|
else:
|
|
result = managementInterface.getAllPrompts(pagination=paginationParams)
|
|
if paginationParams and hasattr(result, 'items'):
|
|
response = {
|
|
"items": _promptsToEnrichedDicts(result.items),
|
|
"pagination": PaginationMetadata(
|
|
currentPage=paginationParams.page,
|
|
pageSize=paginationParams.pageSize,
|
|
totalItems=result.totalItems,
|
|
totalPages=result.totalPages,
|
|
sort=paginationParams.sort,
|
|
filters=paginationParams.filters
|
|
).model_dump(),
|
|
}
|
|
else:
|
|
response = {"items": _promptsToEnrichedDicts(result if isinstance(result, list) else [result]), "pagination": None}
|
|
if viewMeta:
|
|
response["appliedView"] = viewMeta.model_dump()
|
|
return response
|
|
|
|
# Strategy B grouping: load all, filter+sort in-memory, group, then slice
|
|
result = managementInterface.getAllPrompts(pagination=None)
|
|
allItems = _promptsToEnrichedDicts(result if isinstance(result, list) else (result.items if hasattr(result, 'items') else []))
|
|
|
|
if not paginationParams:
|
|
response = {"items": allItems, "pagination": None}
|
|
if viewMeta:
|
|
response["appliedView"] = viewMeta.model_dump()
|
|
return response
|
|
|
|
if paginationParams.filters or paginationParams.sort:
|
|
from modules.interfaces.interfaceDbManagement import ComponentObjects
|
|
comp = ComponentObjects()
|
|
comp.setUserContext(currentUser)
|
|
filtersForGenericApply = paginationParams.filters
|
|
promptSearchValue = None
|
|
if paginationParams.filters:
|
|
promptSearchValue, filtersForGenericApply = _extractSearchAndRemoveFromFilters(paginationParams.filters)
|
|
if promptSearchValue is not None:
|
|
allItems = _applyPromptSearchOnVisibleFields(allItems, promptSearchValue)
|
|
if filtersForGenericApply:
|
|
allItems = comp._applyFilters(allItems, filtersForGenericApply)
|
|
if paginationParams.sort:
|
|
allItems = comp._applySorting(allItems, paginationParams.sort)
|
|
|
|
totalItems = len(allItems)
|
|
totalPages = math.ceil(totalItems / paginationParams.pageSize) if totalItems > 0 else 0
|
|
page_items, groupLayout = buildGroupLayout(allItems, groupByLevels, paginationParams.page, paginationParams.pageSize)
|
|
|
|
response = {
|
|
"items": page_items,
|
|
"pagination": PaginationMetadata(
|
|
currentPage=paginationParams.page,
|
|
pageSize=paginationParams.pageSize,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages,
|
|
sort=paginationParams.sort,
|
|
filters=paginationParams.filters
|
|
).model_dump(),
|
|
}
|
|
if groupLayout:
|
|
response["groupLayout"] = groupLayout.model_dump()
|
|
if viewMeta:
|
|
response["appliedView"] = viewMeta.model_dump()
|
|
return response
|
|
|
|
|
|
@router.post("", response_model=Prompt)
|
|
@limiter.limit("10/minute")
|
|
def create_prompt(
|
|
request: Request,
|
|
prompt: Prompt,
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> Prompt:
|
|
"""Create a new prompt"""
|
|
managementInterface = interfaceDbManagement.getInterface(currentUser)
|
|
|
|
# Create prompt
|
|
newPrompt = managementInterface.createPrompt(prompt)
|
|
|
|
return Prompt(**newPrompt)
|
|
|
|
@router.get("/{promptId}", response_model=Prompt)
|
|
@limiter.limit("30/minute")
|
|
def get_prompt(
|
|
request: Request,
|
|
promptId: str = Path(..., description="ID of the prompt"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> Prompt:
|
|
"""Get a specific prompt"""
|
|
managementInterface = interfaceDbManagement.getInterface(currentUser)
|
|
|
|
# Get prompt
|
|
prompt = managementInterface.getPrompt(promptId)
|
|
if not prompt:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Prompt with ID {promptId} not found"
|
|
)
|
|
|
|
return prompt
|
|
|
|
@router.put("/{promptId}", response_model=Prompt)
|
|
@limiter.limit("10/minute")
|
|
def update_prompt(
|
|
request: Request,
|
|
promptId: str = Path(..., description="ID of the prompt to update"),
|
|
promptData: Dict[str, Any] = Body(...),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> Prompt:
|
|
"""Update an existing prompt (supports partial updates for inline editing)"""
|
|
managementInterface = interfaceDbManagement.getInterface(currentUser)
|
|
|
|
# Check if the prompt exists
|
|
existingPrompt = managementInterface.getPrompt(promptId)
|
|
if not existingPrompt:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Prompt with ID {promptId} not found"
|
|
)
|
|
|
|
# Remove id from update data if present
|
|
update_data = {k: v for k, v in promptData.items() if k != "id"}
|
|
|
|
# Update prompt (ownership check happens in interface)
|
|
try:
|
|
updatedPrompt = managementInterface.updatePrompt(promptId, update_data)
|
|
except PermissionError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=str(e)
|
|
)
|
|
|
|
if not updatedPrompt:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=routeApiMsg("Error updating the prompt")
|
|
)
|
|
|
|
return Prompt(**updatedPrompt)
|
|
|
|
@router.delete("/{promptId}", response_model=Dict[str, Any])
|
|
@limiter.limit("10/minute")
|
|
def delete_prompt(
|
|
request: Request,
|
|
promptId: str = Path(..., description="ID of the prompt to delete"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> Dict[str, Any]:
|
|
"""Delete a prompt"""
|
|
managementInterface = interfaceDbManagement.getInterface(currentUser)
|
|
|
|
# Check if the prompt exists
|
|
existingPrompt = managementInterface.getPrompt(promptId)
|
|
if not existingPrompt:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Prompt with ID {promptId} not found"
|
|
)
|
|
|
|
try:
|
|
success = managementInterface.deletePrompt(promptId)
|
|
except PermissionError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=str(e)
|
|
)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=routeApiMsg("Error deleting the prompt")
|
|
)
|
|
|
|
return {"message": f"Prompt with ID {promptId} successfully deleted"} |