119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
import logging
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.exc import NoResultFound
|
|
from sqlmodel import Session, select
|
|
|
|
from app.model.models import Card, CardCreate, CardUpdate, GroupDB
|
|
from app.services.auth import auth_is_admin
|
|
from app.services.database import add_and_refresh, get_session
|
|
from app.services.scanner import DeleteCard, WriteNewCard
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
|
|
|
|
|
|
def register_card(cardInput: CardCreate):
|
|
key, uid = WriteNewCard()
|
|
if key is None:
|
|
logger.info("No card registered. Check logs!")
|
|
raise HTTPException(
|
|
status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!"
|
|
)
|
|
card = Card(
|
|
group_id=cardInput.group_id,
|
|
key=key,
|
|
name=cardInput.name,
|
|
card_serial=uid,
|
|
enabled=cardInput.enabled,
|
|
)
|
|
return card
|
|
|
|
|
|
@card_router.post("/", response_model=Card)
|
|
def add_card(
|
|
cardInput: CardCreate,
|
|
db: Session = Depends(get_session),
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
try:
|
|
assert (
|
|
db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none()
|
|
is None
|
|
)
|
|
except AssertionError as err:
|
|
raise HTTPException(
|
|
status.HTTP_409_CONFLICT, detail="Name already used!"
|
|
) from err
|
|
try:
|
|
assert (
|
|
db.exec(
|
|
select(GroupDB).where(GroupDB.id == cardInput.group_id)
|
|
).one_or_none()
|
|
is not None
|
|
)
|
|
except AssertionError as err:
|
|
raise HTTPException(
|
|
status.HTTP_404_NOT_FOUND, detail="GroupID not found!"
|
|
) from err
|
|
card = register_card(cardInput)
|
|
return add_and_refresh(db, card)
|
|
|
|
|
|
@card_router.delete("/")
|
|
def del_card(
|
|
*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
|
|
):
|
|
key = DeleteCard()
|
|
logger.info(key)
|
|
try:
|
|
card = db.exec(select(Card).where(Card.key == key)).one()
|
|
except NoResultFound as err:
|
|
logger.info(f"The key:'{key}' was not found in db!")
|
|
raise HTTPException(
|
|
status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Key on card not found in DB. KEY={key}",
|
|
) from err
|
|
db.delete(card)
|
|
db.commit()
|
|
return {"message": "Card deleted successfully"}
|
|
|
|
|
|
@card_router.get("/{group_id}", response_model=list[Card])
|
|
def get_cards(
|
|
*,
|
|
db: Session = Depends(get_session),
|
|
group_id: int,
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
cards = db.exec(select(Card).where(Card.group_id == group_id)).all()
|
|
return cards
|
|
|
|
|
|
@card_router.patch("/{card_id}", response_model=Card)
|
|
def update_card(
|
|
card_id: int,
|
|
cardInput: CardUpdate,
|
|
db: Session = Depends(get_session),
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
db_card = db.get(Card, card_id)
|
|
if db_card is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Card not found!")
|
|
card_data = cardInput.model_dump(exclude_unset=True, exclude_none=True)
|
|
if "group_id" in card_data:
|
|
try:
|
|
assert (
|
|
db.exec(
|
|
select(GroupDB).where(GroupDB.id == cardInput.group_id)
|
|
).one_or_none()
|
|
is not None
|
|
)
|
|
except AssertionError as err:
|
|
raise HTTPException(
|
|
status.HTTP_404_NOT_FOUND, detail="GroupID not found!"
|
|
) from err
|
|
db_card.sqlmodel_update(card_data)
|
|
return add_and_refresh(db, db_card)
|