Spaces:
Sleeping
Sleeping
File size: 4,239 Bytes
57cf043 fd3c8b9 57cf043 fd3c8b9 57cf043 fd3c8b9 57cf043 fd3c8b9 57cf043 fd3c8b9 57cf043 |
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 |
import logging
from datetime import datetime
from typing import Annotated, Optional
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import aliased
from starlette import status
from common import auth
from common.common import configure_logging
from common.exceptions import LogNotFoundException
from components.dbo.models.feedback import Feedback
from components.dbo.models.log import Log
from schemas.log import LogCreate
import common.dependencies as DI
from sqlalchemy.orm import sessionmaker
router = APIRouter(tags=['Logs'])
logger = logging.getLogger(__name__)
configure_logging()
@router.get('/logs', status_code=status.HTTP_200_OK)
async def get_all_logs(
db: Annotated[sessionmaker, Depends(DI.get_db)],
current_user: Annotated[any, Depends(auth.get_current_user)],
date_start: Optional[datetime] = Query(None, alias="date_start"),
date_end: Optional[datetime] = Query(None, alias="date_end")
):
logger.info(f'GET /logs: start')
logger.info(f'GET /logs: start_date={date_start}, end_date={date_end}')
feedback_alias = aliased(Feedback)
query = db.query(Log)
if date_start and date_end:
query = query.filter(Log.dateCreated.between(date_start, date_end))
elif date_start:
query = query.filter(Log.dateCreated >= date_start)
elif date_end:
query = query.filter(Log.dateCreated <= date_end)
query = query.outerjoin(feedback_alias, Log.id == feedback_alias.log_id)
logs_with_feedback = query.all()
combined_logs = []
for log in logs_with_feedback:
if log.feedback:
for feedback in log.feedback:
combined_logs.append(
{
"log_id": log.id,
"llmPrompt": log.llmPrompt,
"llmResponse": log.llmResponse,
"llm_classifier": log.llm_classifier,
"dateCreated": log.dateCreated,
"userRequest": log.userRequest,
"userName": log.userName,
"query_type": log.query_type,
"feedback_id": feedback.feedback_id,
"userComment": feedback.userComment,
"userScore": feedback.userScore,
"manualEstimate": feedback.manualEstimate,
"llmEstimate": feedback.llmEstimate,
}
)
else:
combined_logs.append(
{
"log_id": log.id,
"llmPrompt": log.llmPrompt,
"llmResponse": log.llmResponse,
"llm_classifier": log.llm_classifier,
"dateCreated": log.dateCreated,
"userRequest": log.userRequest,
"userName": log.userName,
"query_type": log.query_type,
"feedback_id": None,
"userComment": None,
"userScore": None,
"manualEstimate": None,
"llmEstimate": None,
}
)
return combined_logs
@router.get('/log/{log_id}', status_code=status.HTTP_200_OK)
async def get_log(db: Annotated[sessionmaker, Depends(DI.get_db)],
current_user: Annotated[any, Depends(auth.get_current_user)], log_id):
log = db.query(Log).filter(Log.id == log_id).first()
if log is None:
raise LogNotFoundException(log_id)
return log
@router.post('/log', status_code=status.HTTP_201_CREATED)
async def create_log(log: LogCreate, db: Annotated[sessionmaker, Depends(DI.get_db)]):
logger.info("Handling POST request to /log")
try:
new_log = Log(
llmPrompt=log.llmPrompt,
llmResponse=log.llmResponse,
llm_classifier=log.llm_classifier,
userRequest=log.userRequest,
userName=log.userName,
)
db.add(new_log)
db.commit()
db.refresh(new_log)
logger.info(f"Successfully created log with ID: {new_log.id}")
return new_log
except Exception as e:
logger.error(f"Error creating log: {str(e)}")
raise e
|