private_gpt/launcher.py

"""FastAPI app creation, logger configuration and main API routes."""
import logging
from typing import Any

from fastapi import Depends, FastAPI, Request, status, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.utils import get_openapi
from injector import Injector
from fastapi import APIRouter
#from fastapi import JSONResponse

from private_gpt.paths import docs_path
from private_gpt.server.chat.chat_router import chat_router
from private_gpt.server.chunks.chunks_router import chunks_router
from private_gpt.server.completions.completions_router import completions_router
from private_gpt.server.embeddings.embeddings_router import embeddings_router
from private_gpt.server.health.health_router import health_router
from private_gpt.server.ingest.ingest_router import ingest_router
from private_gpt.server.utils import authentication
from private_gpt.settings.settings import Settings
from private_gpt.components.llm.llm_component import LLMComponent

#databse for user authentication integration
#from private_gpt.server.utils import models 
#from private_gpt.server.utils.database import engine, SessionLocal
from typing import Annotated
from sqlalchemy.orm import Session
from private_gpt.server.utils.authentication import get_current_user

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2AuthorizationCodeBearer



logger = logging.getLogger(__name__)


def create_app(root_injector: Injector) -> FastAPI:

    # Start the API
    with open(docs_path / "description.md") as description_file:
        description = description_file.read()

        tags_metadata = [
            {
                "name": "Ingestion",
                "description": "High-level APIs covering document ingestion -internally "
                "managing document parsing, splitting,"
                "metadata extraction, embedding generation and storage- and ingested "
                "documents CRUD."
                "Each ingested document is identified by an ID that can be used to filter the "
                "context"
                "used in *Contextual Completions* and *Context Chunks* APIs.",
            },
            {
                "name": "Contextual Completions",
                "description": "High-level APIs covering contextual Chat and Completions. They "
                "follow OpenAI's format, extending it to "
                "allow using the context coming from ingested documents to create the "
                "response. Internally"
                "manage context retrieval, prompt engineering and the response generation.",
            },
            {
                "name": "Context Chunks",
                "description": "Low-level API that given a query return relevant chunks of "
                "text coming from the ingested"
                "documents.",
            },
            {
                "name": "Embeddings",
                "description": "Low-level API to obtain the vector representation of a given "
                "text, using an Embeddings model."
                "Follows OpenAI's embeddings API format.",
            },
            {
                "name": "Health",
                "description": "Simple health API to make sure the server is up and running.",
            },
        ]

        async def bind_injector_to_request(request: Request) -> None:
            request.state.injector = root_injector

        app = FastAPI(dependencies=[Depends(bind_injector_to_request)])
        

        model_router = APIRouter(prefix ='/v1',dependencies=[Depends(get_current_user)])

        @model_router.post("/switch_model")
        async def switch_model(
            new_model: str, current_user: dict = Depends(get_current_user)
        ):
            # Check if the user has the "admin" role
            if "admin" not in current_user.get("role", []):
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail="You are not an admin and cannot use this API.",
                )

            #logic to switch the LLM model based on the user's request
            
            llm_component = root_injector.get(LLMComponent)
            llm_component.switch_model(new_model, settings=settings)

            return {"message": f"Model switched to {new_model}"}

        # Define a new APIRouter for the model_list
        model_list_router = APIRouter(prefix="/v1", dependencies=[Depends(get_current_user)])    

        @model_list_router.get("/models_list", response_model=list[dict])
        async def model_list(current_user: dict = Depends(get_current_user)):
            """
            Get a list of models with their details.
            """
            # In this example, hardcoding some sample model data
            models_data = [
                {"id": 1, "name": "gpt-3.5-turbo", "access": ["user", "admin"]},
                {"id": 2, "name": "gpt-4", "access": ["admin"]},
                # Add more models as needed
            ]

            return models_data

        # @model_router.post("/switch_model")
        # async def switch_model(new_model: str):
        #     # Implement logic to switch the LLM model based on the user's request
        #     # Example: Change the LLM model in LLMComponent based on the new_model parameter
        #     llm_component = root_injector.get(LLMComponent)
        #     llm_component.switch_model(new_model, settings=settings)

        #     return {"message": f"Model switched to {new_model}"}


        def custom_openapi() -> dict[str, Any]:
            if app.openapi_schema:
                return app.openapi_schema
            openapi_schema = get_openapi(
                title="Invenxion-Chatbot",
                description=description,
                version="0.1.0",
                summary="This is a production-ready AI project that allows you to "
                "ask questions to your documents using the power of Large Language "
                "Models (LLMs), even in scenarios without Internet connection. "
                "100% private, no data leaves your execution environment at any point.",
                license_info={
                    "name": "Apache 2.0",
                    "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
                },
                routes=app.routes,
                tags=tags_metadata,
            )
            openapi_schema["info"]["x-logo"] = {
                "url": "https://lh3.googleusercontent.com/drive-viewer"
                "/AK7aPaD_iNlMoTquOBsw4boh4tIYxyEuhz6EtEs8nzq3yNkNAK00xGj"
                "E1KUCmPJSk3TYOjcs6tReG6w_cLu1S7L_gPgT9z52iw=s2560"
            }

            app.openapi_schema = openapi_schema
            return app.openapi_schema

        app.openapi = custom_openapi  # type: ignore[method-assign]

        #models.Base.metadata.create_all(bind=engine)

        # def get_db():
        #     db = SessionLocal()
        #     try:
        #         yield db
        #     finally:
        #         db.close()

        #db_dependency = Annotated[Session, Depends(get_db)]
        #user_dependency = Annotated[dict, Depends(get_current_user)]

        @app.get("/v1/me", status_code=status.HTTP_200_OK)
        async def user(current_user: dict = Depends(get_current_user)):
            if current_user is None:
                raise HTTPException(status_code=401, detail="Authentication Failed")
            return {"User": current_user}

        
        app.include_router(authentication.router)

        app.include_router(completions_router)
        app.include_router(chat_router)
        app.include_router(chunks_router)
        app.include_router(ingest_router)
        app.include_router(embeddings_router)
        app.include_router(health_router)
        app.include_router(model_router)
        app.include_router(model_list_router)

        settings = root_injector.get(Settings)
        if settings.server.cors.enabled:
            logger.debug("Setting up CORS middleware")
            app.add_middleware(
                CORSMiddleware,
                allow_credentials=settings.server.cors.allow_credentials,
                allow_origins=settings.server.cors.allow_origins,
                allow_origin_regex=settings.server.cors.allow_origin_regex,
                allow_methods=settings.server.cors.allow_methods,
                allow_headers=settings.server.cors.allow_headers,
            )

        if settings.ui.enabled:
            logger.debug("Importing the UI module")
            from private_gpt.ui.ui import PrivateGptUi

            ui = root_injector.get(PrivateGptUi)
            ui.mount_in_app(app, settings.ui.path)

        return app