82 lines
No EOL
3 KiB
Python
82 lines
No EOL
3 KiB
Python
from fastapi import APIRouter, HTTPException, Depends, Path, Response
|
|
from typing import List, Dict, Any
|
|
from fastapi import status
|
|
import inspect
|
|
import importlib
|
|
import os
|
|
from pydantic import BaseModel
|
|
|
|
from modules.security.auth import getCurrentActiveUser, getUserContext
|
|
|
|
# Import the attribute definition and helper functions
|
|
from modules.shared.defAttributes import AttributeDefinition, getModelAttributes
|
|
|
|
def getModelClasses() -> Dict[str, Any]:
|
|
"""Dynamically get all model classes from all model modules"""
|
|
modelClasses = {}
|
|
|
|
# Get the interfaces directory path
|
|
# Since we're in modules/routes/, we need to go up one level to modules/ then into interfaces/
|
|
interfaces_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'interfaces')
|
|
|
|
# Find all model files
|
|
for filename in os.listdir(interfaces_dir):
|
|
if filename.endswith('Model.py'):
|
|
# Convert filename to module name (e.g., gatewayModel.py -> gatewayModel)
|
|
module_name = filename[:-3]
|
|
|
|
# Import the module dynamically
|
|
module = importlib.import_module(f'modules.interfaces.{module_name}')
|
|
|
|
# Get all classes from the module
|
|
for name, obj in inspect.getmembers(module):
|
|
if inspect.isclass(obj) and issubclass(obj, BaseModel) and obj != BaseModel:
|
|
modelClasses[name.lower()] = obj
|
|
|
|
return modelClasses
|
|
|
|
# Create a router for the attribute endpoints
|
|
router = APIRouter(
|
|
prefix="/api/attributes",
|
|
tags=["Attributes"],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
@router.get("/{entityType}", response_model=List[AttributeDefinition])
|
|
async def getEntityAttributes(
|
|
entityType: str = Path(..., description="Type of entity (e.g. prompt)"),
|
|
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
|
|
):
|
|
"""
|
|
Retrieves the attribute definitions for a specific entity.
|
|
This can be used for dynamic form generation.
|
|
"""
|
|
# Authentication and user context
|
|
mandateId, userId = await getUserContext(currentUser)
|
|
|
|
# Determine preferred language of the user
|
|
userLanguage = currentUser.get("language", "de")
|
|
|
|
# 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]
|
|
attributes = getModelAttributes(modelClass, userLanguage)
|
|
|
|
# Return only visible attributes
|
|
return [attr for attr in attributes if attr.visible]
|
|
|
|
@router.options("/{entityType}")
|
|
async def optionsEntityAttributes(
|
|
entityType: str = Path(..., description="Type of entity (e.g. prompt)")
|
|
):
|
|
"""Handle OPTIONS request for CORS preflight"""
|
|
return Response(status_code=200) |