methinks I should've done this a lot sooner

ran 'ruff format app'
This commit is contained in:
2026-07-29 01:59:34 +02:00
parent 1ca046ded9
commit 4ce98a2be6
14 changed files with 377 additions and 136 deletions

View File

@@ -12,40 +12,65 @@ from app.services.database import add_and_refresh, get_session
aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"])
@aa_router.post("/", response_model=AccessAuthorizationResponse)
def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate, admin: bool = Depends(auth_is_admin)):
def add_accessauth(
*,
db: Session = Depends(get_session),
aa: AccessAuthorizationCreate,
admin: bool = Depends(auth_is_admin),
):
logger.info(f"Creating accessauth with data: {aa}")
if aa.timetables != []:
timetables = [Timetable.model_validate(t) for t in aa.timetables]
else: timetables = []
else:
timetables = []
if aa.oneshot is not None:
oneshot = OneShotAccess.model_validate(aa.oneshot)
else: oneshot = None
else:
oneshot = None
db_aa = AccessAuthorizationDB(
name=aa.name,
type=aa.type,
is_active=aa.is_active,
timetables=timetables,
oneshot=oneshot
oneshot=oneshot,
)
return add_and_refresh(db, db_aa)
@aa_router.get("/", response_model=list[AccessAuthorizationResponse])
def get_all_accessauths(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
def get_all_accessauths(
db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
):
return db.exec(
select(AccessAuthorizationDB)
.options(selectinload(AccessAuthorizationDB.timetables))
).all()
select(AccessAuthorizationDB).options(
selectinload(AccessAuthorizationDB.timetables)
)
).all()
@aa_router.get("/{aa_id}", response_model=AccessAuthorizationResponse)
def get_one_accessauth(*, db: Session = Depends(get_session), aa_id: int, admin: bool = Depends(auth_is_admin)):
def get_one_accessauth(
*,
db: Session = Depends(get_session),
aa_id: int,
admin: bool = Depends(auth_is_admin),
):
db_aa = db.get(AccessAuthorizationDB, aa_id)
if db_aa is None:
raise HTTPException(status_code=404, detail="AA not found")
return db_aa
@aa_router.put("/assign/{group_id}/{aa_id}", response_model=GroupResponse)
def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int, admin: bool = Depends(auth_is_admin)):
def assign_accessauth(
*,
db: Session = Depends(get_session),
group_id: int,
aa_id: int,
admin: bool = Depends(auth_is_admin),
):
db_group = db.get(GroupDB, group_id)
if db_group is None:
raise HTTPException(status_code=404, detail="Group not found")
@@ -53,12 +78,21 @@ def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_i
if db_aa is None:
raise HTTPException(status_code=404, detail="AA not found")
if db_aa in db_group.accessauths:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group")
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group"
)
db_group.accessauths.append(db_aa)
return add_and_refresh(db, db_group)
@aa_router.put("/unassign/{group_id}/{aa_id}", response_model=GroupResponse)
def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int, admin: bool = Depends(auth_is_admin)):
def unassign_accessauth(
*,
db: Session = Depends(get_session),
group_id: int,
aa_id: int,
admin: bool = Depends(auth_is_admin),
):
db_group = db.get(GroupDB, group_id)
if db_group is None:
raise HTTPException(status_code=404, detail="Group not found")
@@ -70,8 +104,15 @@ def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa
db_group.accessauths.remove(db_aa)
return add_and_refresh(db, db_group)
@aa_router.patch("/{aa_id}", response_model=AccessAuthorizationResponse)
def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: AccessAuthorizationUpdate, admin: bool = Depends(auth_is_admin)):
def change_accessauth(
*,
db: Session = Depends(get_session),
aa_id: int,
aa: AccessAuthorizationUpdate,
admin: bool = Depends(auth_is_admin),
):
db_aa = db.get(AccessAuthorizationDB, aa_id)
if db_aa is None:
raise HTTPException(status_code=404, detail="AccessAuthorization not found")
@@ -88,11 +129,17 @@ def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: Acc
db_aa.sqlmodel_update(aa_data)
return add_and_refresh(db, db_aa)
@aa_router.delete("/{aa_id}")
def delete_accessauth(*, db: Session = Depends(get_session), aa_id: int, admin: bool = Depends(auth_is_admin)):
def delete_accessauth(
*,
db: Session = Depends(get_session),
aa_id: int,
admin: bool = Depends(auth_is_admin),
):
db_aa = db.get(AccessAuthorizationDB, aa_id)
if db_aa is None:
raise HTTPException(status_code=404, detail="AccessAuthorization not found")
db.delete(db_aa)
db.commit()
return {"message": "AccessAuthorization deleted successfully"}
return {"message": "AccessAuthorization deleted successfully"}

View File

@@ -13,58 +13,100 @@ from app.services.scanner import DeleteCard, WriteNewCard
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
def register_card(cardInput: CardCreate):
key, uid = WriteNewCard()
if key == None:
logger.info("No card registered. Check logs!")
raise HTTPException(status.HTTP_417_EXPECTATION_FAILED, detail="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
)
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() == None
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()
== None
)
except AssertionError:
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None
try:
assert (
db.exec(
select(GroupDB).where(GroupDB.id == cardInput.group_id)
).one_or_none()
is not None
)
except AssertionError:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!")
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)):
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:
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. Please tell an admin about this. KEY={key}")
raise HTTPException(
status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Key on card not found in DB. Please tell an admin about this. KEY={key}",
)
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)):
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)):
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
try:
assert (
db.exec(
select(GroupDB).where(GroupDB.id == cardInput.group_id)
).one_or_none()
is not None
)
except AssertionError:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!")
db_card.sqlmodel_update(card_data)
return add_and_refresh(db, db_card)
return add_and_refresh(db, db_card)

View File

@@ -15,20 +15,30 @@ debug_router = APIRouter(
dependencies=[Depends(auth_is_admin)],
)
@debug_router.put("/addcard/")
def add_card_manually(groupid: int, card_key: str, name: str, enabled: bool, db: Session=Depends(get_session)):
def add_card_manually(
groupid: int,
card_key: str,
name: str,
enabled: bool,
db: Session = Depends(get_session),
):
"""Add cards manually (you also have to delete them manually)"""
logger.critical(f"Manual db change: adding a card with key: {card_key} to group: {groupid}")
logger.critical(
f"Manual db change: adding a card with key: {card_key} to group: {groupid}"
)
card = Card(
group_id=groupid,
key=card_key,
name=name,
enabled=enabled,
card_serial="00:00:00:00:00:00:00"
)
card_serial="00:00:00:00:00:00:00",
)
add_and_refresh(db, card)
return card
@debug_router.get("/rmcard/{card_id}")
def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
try:
@@ -39,7 +49,7 @@ def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
db.delete(card)
db.commit()
return {"message": "Card deleted successfully"}
@debug_router.put("/getcards")
def list_all_cards(db: Session = Depends(get_session)):
@@ -49,4 +59,4 @@ def list_all_cards(db: Session = Depends(get_session)):
out = []
for i in cards:
out.append({i.key: i.group.name})
return out
return out

View File

@@ -5,16 +5,19 @@ import app.services.door as doorService
from app.services.auth import auth_is_admin
from app.services.database import get_session
door_router = APIRouter(prefix="/api/v1/door",tags=["Door"])
door_router = APIRouter(prefix="/api/v1/door", tags=["Door"])
@door_router.put("/open")
def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
doorService.opendoor()
@door_router.put("/close")
def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
doorService.closedoor()
@door_router.post("/test")
def test_access(input: str, db: Session = Depends(get_session)):
return doorService.checkAccess(input, db=db)
return doorService.checkAccess(input, db=db)

View File

@@ -1,4 +1,3 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
@@ -8,24 +7,41 @@ from ..services.database import add_and_refresh, get_session
group_router = APIRouter(prefix="/api/v1/groups", tags=["Group"])
@group_router.get("/", response_model=list[GroupResponse])
def get_groups(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
def get_groups(
*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
):
groups = db.exec(select(GroupDB)).all()
return groups
@group_router.post("/", response_model=GroupResponse)
def create_group(*, db: Session = Depends(get_session), group: GroupCreate, admin: bool = Depends(auth_is_admin)):
def create_group(
*,
db: Session = Depends(get_session),
group: GroupCreate,
admin: bool = Depends(auth_is_admin),
):
db_group = GroupDB.model_validate(group)
group = db.exec(select(GroupDB).where(GroupDB.name == db_group.name)).first()
if group is not None:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Group already exists!")
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Group already exists!"
)
return add_and_refresh(db, db_group)
@group_router.delete("/{group_id}")
def delete_group(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)):
def delete_group(
*,
db: Session = Depends(get_session),
group_id: int,
admin: bool = Depends(auth_is_admin),
):
db_group = db.get(GroupDB, group_id)
if db_group is None:
raise HTTPException(status_code=404, detail="Group not found")
db.delete(db_group)
db.commit()
return {"message": "Group deleted successfully"}
return {"message": "Group deleted successfully"}

View File

@@ -12,33 +12,62 @@ from ..services.database import add_and_refresh, get_session
user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
@user_router.post("/", response_model=UserResponse)
def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin: bool = Depends(auth_is_admin)):
def create_user(
*,
db: Session = Depends(get_session),
user: UserCreate,
admin: bool = Depends(auth_is_admin),
):
hashed_password = {"passwordhash": get_password_hash(user.password)}
try: assert db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none() == None
try:
assert (
db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none()
== None
)
except AssertionError:
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
db_user = UserDB.model_validate(user, update=hashed_password)
return add_and_refresh(db, db_user)
@user_router.get("/", response_model=list[UserResponse])
def read_users(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
def read_users(
*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
):
users = db.exec(select(UserDB)).all()
return users
@user_router.get("/current", response_model=UserResponse)
def get_current_user(db: Session = Depends(get_session), user: UserDB = Depends(auth_user)):
def get_current_user(
db: Session = Depends(get_session), user: UserDB = Depends(auth_user)
):
return user
@user_router.get("/{user_id}", response_model=UserResponse)
def read_user(*, db: Session = Depends(get_session), user_id: int, admin: bool = Depends(auth_is_admin)):
def read_user(
*,
db: Session = Depends(get_session),
user_id: int,
admin: bool = Depends(auth_is_admin),
):
db_user = db.get(UserDB, user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
@user_router.patch("/{user_id}", response_model=UserResponse)
def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserUpdate, admin: bool = Depends(auth_is_admin)):
def update_user(
*,
db: Session = Depends(get_session),
user_id: int,
user: UserUpdate,
admin: bool = Depends(auth_is_admin),
):
db_user = db.get(UserDB, user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
@@ -50,12 +79,17 @@ def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserU
db_user.sqlmodel_update(user_data, update=hashed_password)
return add_and_refresh(db, db_user)
@user_router.delete("/{user_id}")
def delete_user(*, db: Session = Depends(get_session), user_id: int, admin: bool = Depends(auth_is_admin)):
def delete_user(
*,
db: Session = Depends(get_session),
user_id: int,
admin: bool = Depends(auth_is_admin),
):
db_user = db.get(UserDB, user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
db.delete(db_user)
db.commit()
return {"message": "User deleted successfully"}