from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.api.deps import get_db from app.api.permissions import require_update from app.crud.metadata import document_metadata_value, metadata_field from app.models import KnowledgeBase, User from app.schemas.metadata import ( FieldReorder, MetadataFieldCreate, MetadataFieldResponse, MetadataFieldUpdate, ) from common_logging import get_logger logger = get_logger(__name__) router = APIRouter(tags=["metadata-fields"]) def check_knowledge_base_access( db: Session, knowledge_base_id: int, current_user: User ) -> KnowledgeBase: kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == knowledge_base_id).first() if not kb: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Knowledge base not found" ) return kb @router.get("/bases/{base_id}/metadata-fields", response_model=list[MetadataFieldResponse]) def get_metadata_fields( base_id: int, db: Session = Depends(get_db), current_user: User = Depends(require_update("knowledge_bases")), ): check_knowledge_base_access(db, base_id, current_user) system_fields = [ MetadataFieldResponse( id=0, knowledge_base_id=base_id, field_name="Document Name", field_key="document_name", field_type="text", field_options=None, default_value=None, is_required=False, sort_order=-5, description="Document name (auto-populated)", created_at=None, updated_at=None, is_system=True, ), MetadataFieldResponse( id=0, knowledge_base_id=base_id, field_name="Uploader", field_key="uploader", field_type="text", field_options=None, default_value=None, is_required=False, sort_order=-4, description="User who uploaded the document (auto-populated)", created_at=None, updated_at=None, is_system=True, ), MetadataFieldResponse( id=0, knowledge_base_id=base_id, field_name="Upload Date", field_key="upload_date", field_type="date", field_options=None, default_value=None, is_required=False, sort_order=-3, description="Date when document was uploaded (auto-populated)", created_at=None, updated_at=None, is_system=True, ), MetadataFieldResponse( id=0, knowledge_base_id=base_id, field_name="Last Update Date", field_key="last_update_date", field_type="date", field_options=None, default_value=None, is_required=False, sort_order=-2, description="Date when document was last updated (auto-populated)", created_at=None, updated_at=None, is_system=True, ), MetadataFieldResponse( id=0, knowledge_base_id=base_id, field_name="Source", field_key="source", field_type="text", field_options=None, default_value=None, is_required=False, sort_order=-1, description="Document source (auto-populated)", created_at=None, updated_at=None, is_system=True, ), ] custom_fields = metadata_field.get_by_knowledge_base(db, knowledge_base_id=base_id) return system_fields + custom_fields @router.post("/bases/{base_id}/metadata-fields", response_model=MetadataFieldResponse) def create_metadata_field( base_id: int, field_in: MetadataFieldCreate, db: Session = Depends(get_db), current_user: User = Depends(require_update("knowledge_bases")), ): check_knowledge_base_access(db, base_id, current_user) existing = metadata_field.get_by_field_key( db, knowledge_base_id=base_id, field_key=field_in.field_key ) if existing: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Field key '{field_in.field_key}' already exists in this knowledge base", ) valid_types = ["text", "textarea", "number", "date", "select", "multiselect"] if field_in.field_type not in valid_types: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid field_type. Must be one of: {', '.join(valid_types)}", ) if field_in.field_type in ["select", "multiselect"]: if not field_in.field_options or len(field_in.field_options) == 0: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="field_options is required for select/multiselect types", ) logger.bind(field_key=field_in.field_key, base_id=base_id).info("Creating metadata field") field = metadata_field.create_with_knowledge_base( db, obj_in=field_in, knowledge_base_id=base_id ) return field @router.put("/metadata-fields/{field_id}", response_model=MetadataFieldResponse) def update_metadata_field( field_id: int, field_in: MetadataFieldUpdate, db: Session = Depends(get_db), current_user: User = Depends(require_update("knowledge_bases")), ): field = metadata_field.get(db, id=field_id) if not field: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Metadata field not found" ) check_knowledge_base_access(db, field.knowledge_base_id, current_user) if field_in.field_type: valid_types = ["text", "textarea", "number", "date", "select", "multiselect"] if field_in.field_type not in valid_types: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid field_type. Must be one of: {', '.join(valid_types)}", ) field = metadata_field.update(db, db_obj=field, obj_in=field_in) return field @router.delete("/metadata-fields/{field_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_metadata_field( field_id: int, db: Session = Depends(get_db), current_user: User = Depends(require_update("knowledge_bases")), ): try: field = metadata_field.get(db, id=field_id) if not field: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Metadata field not found" ) check_knowledge_base_access(db, field.knowledge_base_id, current_user) logger.bind(field_id=field_id).info("Deleting document values for field") document_metadata_value.delete_by_field(db, field_id=field_id) logger.bind(field_id=field_id).info("Deleting metadata field") metadata_field.delete(db, id=field_id) return None except Exception as e: logger.bind(field_id=field_id).opt(exception=True).error("Error deleting metadata field") raise @router.post("/metadata-fields/batch-reorder", status_code=status.HTTP_200_OK) def batch_reorder_fields( reorder_data: list[FieldReorder], db: Session = Depends(get_db), current_user: User = Depends(require_update("knowledge_bases")), ): if not reorder_data: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="reorder_data cannot be empty" ) for item in reorder_data: field = metadata_field.get(db, id=item.field_id) if not field: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Metadata field {item.field_id} not found", ) check_knowledge_base_access(db, field.knowledge_base_id, current_user) reorder_dict = [ {"field_id": item.field_id, "sort_order": item.sort_order} for item in reorder_data ] metadata_field.batch_reorder(db, reorder_data=reorder_dict) return {"message": "Fields reordered successfully"}