server.py

"""
 /*************************************************************************
 * 
 * CONFIDENTIAL
 * __________________
 * 
 *  Copyright (2023-2025) AI Labs, IronOne Technologies, LLC
 *  All Rights Reserved
 *
 *  Author  : Theekshana Samaradiwakara
 *  Description :Python Backend API to chat with private data  
 *  CreatedDate : 14/11/2023
 *  LastModifiedDate : 15/10/2024
 *************************************************************************/
"""

import os
import time
import sys
import logging
import datetime
import uvicorn
from dotenv import load_dotenv

from fastapi import FastAPI, APIRouter, HTTPException, status
from fastapi import HTTPException, status
from fastapi.middleware.cors import CORSMiddleware

from schema import UserQuery, ResponseModel, Document, LoginRequest, UserModel
from controller import get_QA_Answers, get_avaliable_models


def filer():
    return "logs/log"
    # today = datetime.datetime.today()
    # log_filename = f"logs/{today.year}-{today.month:02d}-{today.day:02d}.log"
    # return log_filename


file_handler = logging.FileHandler(filer())
# file_handler = logging.handlers.TimedRotatingFileHandler(filer(),when="D")
file_handler.setLevel(logging.INFO)

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)s (%(name)s) : %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[file_handler],
    force=True,
)

logger = logging.getLogger(__name__)

load_dotenv()
host = os.environ.get("APP_HOST")
port = int(os.environ.get("APP_PORT"))


class ChatAPI:

    def __init__(self):
        self.router = APIRouter()
        self.router.add_api_route("/api/v1/health", self.hello, methods=["GET"])
        self.router.add_api_route("/api/v1/models", self.avaliable_models, methods=["GET"])
        self.router.add_api_route(
            "/api/v1/login", self.login, methods=["POST"], response_model=UserModel
        )
        self.router.add_api_route("/api/v1/chat", self.chat, methods=["POST"])

    async def hello(self):
        return "Hello there!"

    async def avaliable_models(self):
        logger.info("getting avaliable models")
        models = get_avaliable_models()

        if not models:
            logger.exception("models not found")
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="models not found"
            )

        return models

    async def login(self, login_request: LoginRequest):
        logger.info(f"username password: {login_request} ")
        # Dummy user data for demonstration (normally, you'd use a database)
        dummy_users_db = {
            "john_doe": {
                "userId": 1,
                "firstName": "John",
                "lastName": "Doe",
                "userName": "john_doe",
                "password": "password",  # Normally, passwords would be hashed and stored securely
                "token": "dummy_token_123",  # In a real scenario, this would be a JWT or another kind of token
            }
        }
        # Fetch user by username
        # user = dummy_users_db.get(login_request.username)
        user = dummy_users_db.get("john_doe")
        # Validate user credentials
        if not user or user["password"] != login_request.password:
            raise HTTPException(status_code=401, detail="Invalid username or password")

        # Return the user model without the password
        return UserModel(
            userId=user["userId"],
            firstName=user["firstName"],
            lastName=user["lastName"],
            userName=user["userName"],
            token=user["token"],
        )

    async def chat(
        self, userQuery: UserQuery
    ):  #:UserQuery):# -> ResponseModel: #chat: QueryModel): # -> ResponseModel:
        """Makes query to doc store via Langchain pipeline.

        :param chat.: question, model, dataset location, history of the chat.
        :type chat: QueryModel
        """
        logger.info(f"userQuery: {userQuery} ")

        try:
            start = time.time()
            res = get_QA_Answers(userQuery)
            logger.info(
                f"--------------------------  answer: {res} -------------------------- "
            )
            # return res
            end = time.time()
            logger.info(
                f"-------------------------- Server process (took {round(end - start, 2)} s.) \n: {res}"
            )
            print(
                f" \n -------------------------- Server process (took {round(end - start, 2)} s.) -------------------------  \n"
            )
            return res

        except HTTPException as e:
            logger.exception(e)
            raise e

        except Exception as e:
            logger.exception(e)
            raise HTTPException(status_code=400, detail=f"Error : {e}")


# initialize API
app = FastAPI(title="Boardpac chatbot API")
api = ChatAPI()
app.include_router(api.router)

# origins = ['http://localhost:8000','http://192.168.10.100:8000']

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

if __name__ == "__main__":

    host = "0.0.0.0"
    port = 8000

    # config = uvicorn.Config("server:app",host=host, port=port, log_config= logging.basicConfig())
    config = uvicorn.Config("server:app", host=host, port=port)
    server = uvicorn.Server(config)
    server.run()
    # uvicorn.run(app)