GLiNER_Inference_Server/gliner_inference_server/helpers.py

from fastapi import HTTPException, status
from typing import List, Optional, Union, Dict, Any, Tuple
import os


MAX_TEXT_LENGTH = int(os.getenv("MAX_TEXT_LENGTH", "1000"))


def validate_single_or_batch(text: Optional[str], texts: Optional[List[str]]):
    """Validate that either text or texts is provided, but not both"""
    if text is None and texts is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={
                "error": {
                    "code": "INVALID_INPUT",
                    "message": "Either 'text' or 'texts' must be provided",
                    "details": {}
                }
            }
        )
    if text is not None and texts is not None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={
                "error": {
                    "code": "INVALID_INPUT",
                    "message": "Provide either 'text' or 'texts', not both",
                    "details": {}
                }
            }
        )

    if text == "" and texts is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={
                "error": {
                    "code": "INVALID_INPUT",
                    "message": "'text' cannot be empty or only whitespace",
                    "details": {}
                }
            }
        )

    if texts is not None and len(texts) == 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={
                "error": {
                    "code": "INVALID_INPUT",
                    "message": "Texts list cannot be empty",
                    "details": {}
                }
            }
        )

    if texts is not None and len(texts) != 0:
        # Check for empty strings within the list
        for i, t in enumerate(texts):
            if not isinstance(t, str) or not t.strip():
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail={
                        "error": {
                            "code": "INVALID_INPUT",
                            "message": f"Text item at index {i} in 'texts' cannot be empty or only whitespace",
                            "details": {}
                        }
                    }
                )

    return text if text is not None else texts


def validate_prompt_and_text_length(prompt: Optional[str], text: Optional[str], texts: Optional[List[str]]):
    """Validate that combined prompt and text length doesn't exceed maximum"""
    prompt_length = len(prompt) if prompt else 0

    if text:
        combined_length = prompt_length + len(text)
        if combined_length > MAX_TEXT_LENGTH:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail={
                    "error": {
                        "code": "INVALID_INPUT",
                        "message": f"Combined prompt and text length ({combined_length}) exceeds maximum length of {MAX_TEXT_LENGTH}",
                        "details": {
                            "prompt_length": prompt_length,
                            "text_length": len(text),
                            "max_length": MAX_TEXT_LENGTH
                        }
                    }
                }
            )

    if texts:
        for i, txt in enumerate(texts):
            combined_length = prompt_length + len(txt)
            if combined_length > MAX_TEXT_LENGTH:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail={
                        "error": {
                            "code": "INVALID_INPUT",
                            "message": f"Combined prompt and texts[{i}] length ({combined_length}) exceeds maximum length of {MAX_TEXT_LENGTH}",
                            "details": {
                                "prompt_length": prompt_length,
                                "text_index": i,
                                "text_length": len(txt),
                                "max_length": MAX_TEXT_LENGTH
                            }
                        }
                    }
                )

def handle_error(e: Exception):
    """Convert exceptions to consistent error format"""
    if isinstance(e, HTTPException):
        raise e

    if isinstance(e, ValueError):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={
                "error": {
                    "code": "INVALID_INPUT",
                    "message": str(e),
                    "details": {}
                }
            }
        )

    # Internal server error
    raise HTTPException(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        detail={
            "error": {
                "code": "INTERNAL_ERROR",
                "message": "An internal error occurred while processing your request",
                "details": {}
            }
        }
    )

def add_prompt_to_text(text: str, prompt: Optional[str]):
    if prompt in [None, ""]:
        return text
    return prompt + text