mirror of
https://github.com/onyx-dot-app/onyx.git
synced 2026-10-22 11:22:49 -05:00
136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
from fastapi import APIRouter
|
|
from fastapi import Depends
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from danswer.auth.schemas import UserRole
|
|
from danswer.auth.users import current_admin_user
|
|
from danswer.auth.users import current_user
|
|
from danswer.db.credentials import create_credential
|
|
from danswer.db.credentials import delete_credential
|
|
from danswer.db.credentials import fetch_credential_by_id
|
|
from danswer.db.credentials import fetch_credentials
|
|
from danswer.db.credentials import update_credential
|
|
from danswer.db.engine import get_session
|
|
from danswer.db.models import User
|
|
from danswer.server.models import CredentialBase
|
|
from danswer.server.models import CredentialSnapshot
|
|
from danswer.server.models import ObjectCreationIdResponse
|
|
from danswer.server.models import StatusResponse
|
|
|
|
|
|
router = APIRouter(prefix="/manage")
|
|
|
|
|
|
"""Admin-only endpoints"""
|
|
|
|
|
|
@router.get("/admin/credential")
|
|
def list_credentials_admin(
|
|
user: User = Depends(current_admin_user),
|
|
db_session: Session = Depends(get_session),
|
|
) -> list[CredentialSnapshot]:
|
|
"""Lists all public credentials"""
|
|
credentials = fetch_credentials(db_session=db_session, user=user)
|
|
return [
|
|
CredentialSnapshot.from_credential_db_model(credential)
|
|
for credential in credentials
|
|
]
|
|
|
|
|
|
@router.delete("/admin/credential/{credential_id}")
|
|
def delete_credential_by_id_admin(
|
|
credential_id: int,
|
|
_: User = Depends(current_admin_user),
|
|
db_session: Session = Depends(get_session),
|
|
) -> StatusResponse:
|
|
"""Same as the user endpoint, but can delete any credential (not just the user's own)"""
|
|
delete_credential(db_session=db_session, credential_id=credential_id, user=None)
|
|
return StatusResponse(
|
|
success=True, message="Credential deleted successfully", data=credential_id
|
|
)
|
|
|
|
|
|
"""Endpoints for all"""
|
|
|
|
|
|
@router.get("/credential")
|
|
def list_credentials(
|
|
user: User | None = Depends(current_user),
|
|
db_session: Session = Depends(get_session),
|
|
) -> list[CredentialSnapshot]:
|
|
credentials = fetch_credentials(db_session=db_session, user=user)
|
|
return [
|
|
CredentialSnapshot.from_credential_db_model(credential)
|
|
for credential in credentials
|
|
]
|
|
|
|
|
|
@router.post("/credential")
|
|
def create_credential_from_model(
|
|
credential_info: CredentialBase,
|
|
user: User | None = Depends(current_user),
|
|
db_session: Session = Depends(get_session),
|
|
) -> ObjectCreationIdResponse:
|
|
if user and user.role != UserRole.ADMIN:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Non-admin cannot create admin credential",
|
|
)
|
|
|
|
return create_credential(credential_info, user, db_session)
|
|
|
|
|
|
@router.get("/credential/{credential_id}")
|
|
def get_credential_by_id(
|
|
credential_id: int,
|
|
user: User = Depends(current_user),
|
|
db_session: Session = Depends(get_session),
|
|
) -> CredentialSnapshot | StatusResponse[int]:
|
|
credential = fetch_credential_by_id(credential_id, user, db_session)
|
|
if credential is None:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail=f"Credential {credential_id} does not exist or does not belong to user",
|
|
)
|
|
|
|
return CredentialSnapshot.from_credential_db_model(credential)
|
|
|
|
|
|
@router.patch("/credential/{credential_id}")
|
|
def update_credential_from_model(
|
|
credential_id: int,
|
|
credential_data: CredentialBase,
|
|
user: User = Depends(current_user),
|
|
db_session: Session = Depends(get_session),
|
|
) -> CredentialSnapshot | StatusResponse[int]:
|
|
updated_credential = update_credential(
|
|
credential_id, credential_data, user, db_session
|
|
)
|
|
if updated_credential is None:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail=f"Credential {credential_id} does not exist or does not belong to user",
|
|
)
|
|
|
|
return CredentialSnapshot(
|
|
id=updated_credential.id,
|
|
credential_json=updated_credential.credential_json,
|
|
user_id=updated_credential.user_id,
|
|
is_admin=updated_credential.is_admin,
|
|
time_created=updated_credential.time_created,
|
|
time_updated=updated_credential.time_updated,
|
|
)
|
|
|
|
|
|
@router.delete("/credential/{credential_id}")
|
|
def delete_credential_by_id(
|
|
credential_id: int,
|
|
user: User = Depends(current_user),
|
|
db_session: Session = Depends(get_session),
|
|
) -> StatusResponse:
|
|
delete_credential(credential_id, user, db_session)
|
|
return StatusResponse(
|
|
success=True, message="Credential deleted successfully", data=credential_id
|
|
)
|