88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
"""
|
|
/dn endpoints.
|
|
"""
|
|
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..db import Connection
|
|
from ..db.schemas import DistinguishedName, DistinguishedNameCreate, User
|
|
from ._common import Responses, get_current_user_if_admin_or_self
|
|
|
|
router = APIRouter(prefix="/dn")
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
responses={
|
|
status.HTTP_200_OK: Responses.OK,
|
|
status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
|
|
status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
|
|
status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
|
|
status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
|
|
status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
|
|
},
|
|
)
|
|
async def add_distinguished_name(
|
|
user_name: str,
|
|
distinguished_name: DistinguishedNameCreate,
|
|
_: User = Depends(get_current_user_if_admin_or_self),
|
|
db: Session | None = Depends(Connection.get),
|
|
):
|
|
"""
|
|
POST ./: Create a new distinguished name in the database.
|
|
"""
|
|
|
|
owner = User.from_db(
|
|
db=db,
|
|
name=user_name,
|
|
)
|
|
|
|
# fail if owner doesn't exist
|
|
if owner is None:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
|
|
|
|
# actually create the new user
|
|
new_dn = DistinguishedName.create(
|
|
db=db,
|
|
dn=distinguished_name,
|
|
owner=owner,
|
|
)
|
|
|
|
# fail if creation was unsuccessful
|
|
if new_dn is None:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
|
|
|
|
# return the created user on success
|
|
return new_dn
|
|
|
|
|
|
# @router.delete(
|
|
# "",
|
|
# responses={
|
|
# status.HTTP_200_OK: Responses.OK,
|
|
# status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
|
|
# status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
|
|
# status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
|
|
# status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
|
|
# },
|
|
# )
|
|
# async def remove_distinguished_name(
|
|
# user_name: str,
|
|
# _: User = Depends(get_current_user_if_admin),
|
|
# db: Session | None = Depends(Connection.get),
|
|
# ):
|
|
# """
|
|
# DELETE ./{user_name}: Remove a user from the database.
|
|
# """
|
|
|
|
# # get the user
|
|
# user = User.from_db(
|
|
# db=db,
|
|
# name=user_name,
|
|
# )
|
|
|
|
# # fail if deletion was unsuccessful
|
|
# if user is None or not user.delete(db):
|
|
# raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
|