82 lines
No EOL
3 KiB
Python
82 lines
No EOL
3 KiB
Python
from fastapi import APIRouter, HTTPException, Path, Response, Request
|
|
from fastapi import status
|
|
import logging
|
|
|
|
# Import auth module
|
|
from modules.security.auth import limiter
|
|
|
|
# Import the attribute definition and helper functions
|
|
from modules.shared.attributeUtils import getModelClasses, getModelAttributeDefinitions, AttributeResponse, AttributeDefinition
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Create a router for the attribute endpoints
|
|
router = APIRouter(
|
|
prefix="/api/attributes",
|
|
tags=["Attributes"],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
@router.get("/{entityType}", response_model=AttributeResponse)
|
|
@limiter.limit("30/minute")
|
|
async def get_entity_attributes(
|
|
request: Request,
|
|
entityType: str = Path(..., description="Type of entity (e.g. prompt)")
|
|
) -> AttributeResponse:
|
|
"""
|
|
Retrieves the attribute definitions for a specific entity.
|
|
This can be used for dynamic form generation.
|
|
|
|
Parameters:
|
|
- entityType: The type of entity to get attributes for (e.g., 'user', 'prompt')
|
|
|
|
Returns:
|
|
- A list of attribute definitions that can be used to generate forms
|
|
"""
|
|
# Get model classes dynamically
|
|
modelClasses = getModelClasses()
|
|
|
|
# Check if entity type is known
|
|
if entityType not in modelClasses:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Entity type '{entityType}' not found."
|
|
)
|
|
|
|
# Get model class and derive attributes from it
|
|
modelClass = modelClasses[entityType]
|
|
try:
|
|
attribute_defs = getModelAttributeDefinitions(modelClass)
|
|
except Exception as e:
|
|
logger.error(f"Error getting attribute definitions for {entityType}: {str(e)}", exc_info=True)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error getting attribute definitions for {entityType}: {str(e)}"
|
|
)
|
|
|
|
# Convert dictionary attributes to AttributeDefinition objects
|
|
attribute_definitions = []
|
|
try:
|
|
for attr in attribute_defs["attributes"]:
|
|
if isinstance(attr, dict) and attr.get('visible', True):
|
|
attribute_definitions.append(AttributeDefinition(**attr))
|
|
elif hasattr(attr, 'visible') and attr.visible:
|
|
attribute_definitions.append(attr)
|
|
except Exception as e:
|
|
logger.error(f"Error converting attribute definitions for {entityType}: {str(e)}", exc_info=True)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error converting attribute definitions for {entityType}: {str(e)}"
|
|
)
|
|
|
|
return AttributeResponse(attributes=attribute_definitions)
|
|
|
|
@router.options("/{entityType}")
|
|
@limiter.limit("60/minute")
|
|
async def options_entity_attributes(
|
|
request: Request,
|
|
entityType: str = Path(..., description="Type of entity (e.g. prompt)")
|
|
) -> Response:
|
|
"""Handle OPTIONS request for CORS preflight"""
|
|
return Response(status_code=200) |