Spaces:
Running
Running
File size: 5,192 Bytes
c2ea5ed bcbd2ec c2ea5ed bcbd2ec c2ea5ed bcbd2ec c2ea5ed bcbd2ec c2ea5ed bcbd2ec c2ea5ed |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
"""
Methods API Router
This module provides API endpoints for managing knowledge extraction methods.
"""
from typing import Dict, Any, List, Optional
from fastapi import APIRouter, HTTPException
from fastapi.responses import JSONResponse
from backend.services.method_service import get_method_service
router = APIRouter(prefix="/api/methods", tags=["methods"])
# Get the method service instance
method_service = get_method_service()
@router.get("/available")
async def get_available_methods(method_type: Optional[str] = None, schema_type: Optional[str] = None) -> Dict[str, Any]:
"""
Get all available knowledge extraction methods with their metadata.
Args:
method_type: Optional filter by method type (production/baseline)
schema_type: Optional filter by schema type (reference_based/direct_based)
Returns:
Dictionary of method names to method information
"""
try:
methods = method_service.get_available_methods()
# Apply filters if specified
if method_type:
methods = {k: v for k, v in methods.items() if v["method_type"] == method_type}
if schema_type:
methods = {k: v for k, v in methods.items() if v["schema_type"] == schema_type}
return {
"status": "success",
"methods": methods,
"count": len(methods),
"filters": {
"method_type": method_type,
"schema_type": schema_type
}
}
except Exception as e:
raise HTTPException(status_code=500, detail="An internal error occurred while retrieving methods")
# Removed separate production/baseline endpoints - use unified /available endpoint with filtering
@router.get("/default")
async def get_default_method() -> Dict[str, Any]:
"""
Get the default method configuration.
Returns:
Default method name and information
"""
try:
default_method = method_service.get_default_method()
method_info = method_service.get_method_info(default_method)
return {
"status": "success",
"default_method": default_method,
"method_info": method_info
}
except Exception as e:
raise HTTPException(status_code=500, detail="An internal error occurred while retrieving default method")
@router.get("/{method_name}")
async def get_method_info(method_name: str) -> Dict[str, Any]:
"""
Get detailed information about a specific method.
Args:
method_name: Name of the method to retrieve
Returns:
Method information including capabilities and requirements
"""
try:
method_info = method_service.get_method_info(method_name)
if not method_info:
raise HTTPException(
status_code=404,
detail=f"Method '{method_name}' not found"
)
return {
"status": "success",
"method": method_info
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail="An internal error occurred while retrieving method info")
@router.post("/{method_name}/validate")
async def validate_method(method_name: str) -> Dict[str, Any]:
"""
Validate a method name and return validation result.
Args:
method_name: Name of the method to validate
Returns:
Validation result with method information if valid
"""
try:
validation_result = method_service.validate_method(method_name)
if not validation_result["valid"]:
return JSONResponse(
status_code=400,
content={
"status": "invalid",
"error": validation_result["error"]
}
)
return {
"status": "valid",
"method_info": validation_result["method_info"]
}
except Exception as e:
raise HTTPException(status_code=500, detail="An internal error occurred while validating method")
@router.get("/{method_name}/schema-compatibility")
async def get_method_schema_compatibility(method_name: str) -> Dict[str, Any]:
"""
Get schema compatibility information for a method.
Args:
method_name: Name of the method
Returns:
Schema compatibility information
"""
try:
compatibility_info = method_service.get_method_schema_compatibility(method_name)
if "error" in compatibility_info:
raise HTTPException(
status_code=404,
detail=compatibility_info["error"]
)
return {
"status": "success",
"compatibility": compatibility_info
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail="An internal error occurred while retrieving compatibility info")
# Removed separate filter endpoints - use unified /available endpoint with query parameters |