[Backend] API 키 관련

API키 관련 기능

유효성 검증

현재는 사용자가 입력한 API키가 그대로 입력됩니다.
그러나 이 API키가 존재하는 않는 키일 수도 있고,
API키 자체는 존재하지만 사용 불가능한 API키일 수도 있습니다.

이런 상황을 방지하기 위하여,
사용자가 API 키를 추가할 때 유효한 API 키인지 검증하는 과정을 추가했습니다.

검증 함수

async def verify_api_key(provider: str, api_key: str, model: str = None):
    try:
        llm = get_llm(provider=provider, model=model, api_key=api_key)

        # 간단한 프롬프트
        prompt = ChatPromptTemplate.from_messages([
            ("user", "Say hello")
        ])

        chain = prompt | llm

        # 실제 요청 (최소 토큰 요청)
        response = await chain.ainvoke({})

        if not response:
            raise ValueError("Empty response received.")

    except Exception as e:
        # 예외 발생하면 키가 유효하지 않다고 판단
        raise ValueError(f"API Key 검증 실패: {str(e)}")

    return {"message": "API Key is valid"}

LLM에 테스트를 위해 간단한 프롬프트를 요청합니다.
이 프롬프트의 결과로 어떠한 답변이 나온다면 그것이 유효한 API키라는 것을 의미하고,
그렇지 않다면 API키가 유효하지 않다는 것을 의미합니다.
이 코드를 테스트 코드를 통해 API키가 입력되면 해당 API키의 유효성을 검증합니다.

엔드포인트

@user_router.post("/AddNewAPIkey", response_model = AddNewAPIkeyResponse)
async def add_new_apikey(request: AddNewAPIkeyRequest, db : Session = Depends(get_db)):
    api_key = request.api_key
    provider_id = request.provider_id
    provider_name = request.provider_name
    usage_limit = request.usage_limit
    usage_count = request.usage_count
    user_id = request.user.user_id
    try:
        verify_api_key(provider=provider_name, api_key=api_key)
    except Exception as e:
        print(f"error occured : {e}")
        return JSONResponse(content={"message": f"오류 발생"}, status_code=500)
    try :
        add_apikey(db=db, api_key=api_key, provider_id=provider_id, provider_name=provider_name,
                   usage_limit=usage_limit, usage_count=usage_count, user_id=user_id)
        return JSONResponse(content={'message': 'API 키가 정상적으로 추가되었습니다.'}, status_code=200)
    except Exception as e:
        return JSONResponse(content={"message": f"오류 발생 : {str(e)}"}, status_code=500)

API 키 삭제

API 키 생성 후 이후에 삭제하는 기능 추가했습니다.

스키마

class DeleteKeyRequest(BaseModel):
    id : int

class DeleteKeyResponse(BaseModel):
    message : str

API키의 ID를 입력받고 메세지를 반환합니다.

CRUD 함수

def delete_apikey(db: Session, key_id : int):
    api_key = db.query(ApiKey).filter(ApiKey.id == key_id).first()
    if not api_key:
        raise ValueError("해당 API Key를 찾을 수 없습니다.")

    db.delete(api_key)
    db.commit()

엔드포인트

@user_router.post("/DeleteAPIKey", response_model=DeleteKeyResponse)
async def delete_api_key(request: DeleteKeyRequest, db: Session = Depends(get_db)):
    key_id = request.id
    try :
        delete_apikey(db = db, key_id = key_id)
        return JSONResponse(content={'message': 'API 키가 삭제되었습니다.'}, status_code=500)
    except Exception as e:
        return JSONResponse(content={'message' : f'에러 발생 : {e}'}, status_code=500)

API키 수정

기존에 추가했던 API 키 내용을 추가하는 코드를 추가했습니다.

스키마

class ChangeKeyreqiest(BaseModel):
    id : int
    api_key : str

class ChangeKeyResponse(BaseModel):
    message : str

이번에는 API 키까지 입력으로 받습니다.

CRUD 함수

def change_apikey(db: Session, key_id : int, api_key : str):
    key = db.query(ApiKey).filter(ApiKey.id == key_id).first()
    if key:
        key.api_key = api_key
    db.commit()
    db.refresh(key)

ID를 기반으로 조회하여 API 키를 입력대로 수정하는 키입니다.

엔드포인트

@user_router.post("/ChangeAPIKey", response_model=ChangeKeyResponse)
async def change_key_endpoint(request: ChangeKeyrequest, db: Session = Depends(get_db)):
    key_id = request.id
    api_key = request.api_key
    try :
        change_apikey(db = db, key_id = key_id, api_key=api_key)
        return JSONResponse(content={'message': 'API 키가 변경되었습니다..'}, status_code=200)
    except Exception as e:
        return JSONResponse(content={'message' : f'에러 발생 : {e}'}, status_code=500)