[lint] fix warnings not autofixed by ruff
This commit is contained in:
@@ -1,15 +1,24 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import selectinload
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.model.models import *
|
||||
from app.model.models import (
|
||||
AccessAuthorizationCreate,
|
||||
AccessAuthorizationResponse,
|
||||
Timetable,
|
||||
AccessAuthorizationDB,
|
||||
OneShotAccess,
|
||||
GroupResponse,
|
||||
GroupDB,
|
||||
AccessAuthorizationUpdate,
|
||||
)
|
||||
from app.services.auth import auth_is_admin
|
||||
from app.services.database import add_and_refresh, get_session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"])
|
||||
|
||||
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.exc import NoResultFound
|
||||
from sqlmodel import Session, select
|
||||
@@ -11,12 +8,14 @@ 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 == None:
|
||||
if key is None:
|
||||
logger.info("No card registered. Check logs!")
|
||||
raise HTTPException(
|
||||
status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!"
|
||||
@@ -40,7 +39,7 @@ def add_card(
|
||||
try:
|
||||
assert (
|
||||
db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none()
|
||||
== None
|
||||
is None
|
||||
)
|
||||
except AssertionError:
|
||||
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
|
||||
|
||||
@@ -1,14 +1,14 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy import exc
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.model.models import *
|
||||
from app.model.models import Card
|
||||
from app.services.auth import auth_is_admin
|
||||
from app.services.database import add_and_refresh, get_session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
debug_router = APIRouter(
|
||||
prefix="/api/v1/debug",
|
||||
tags=["Debug items - maybe dont show this in UI"],
|
||||
|
||||
@@ -14,7 +14,9 @@ def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_
|
||||
|
||||
|
||||
@door_router.put("/close")
|
||||
def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
|
||||
def close_door(
|
||||
db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
|
||||
):
|
||||
doorService.closedoor()
|
||||
|
||||
|
||||
|
||||
@@ -1,14 +1,13 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from ..model.models import UserCreate, UserDB, UserResponse, UserUpdate
|
||||
from ..services.auth import auth_is_admin, get_password_hash
|
||||
from ..services.auth import get_current_user as auth_user
|
||||
from ..services.database import add_and_refresh, get_session
|
||||
from app.model.models import UserCreate, UserDB, UserResponse, UserUpdate
|
||||
from app.services.auth import auth_is_admin, get_password_hash
|
||||
from app.services.auth import get_current_user as auth_user
|
||||
from app.services.database import add_and_refresh, get_session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
|
||||
|
||||
@@ -24,7 +23,7 @@ def create_user(
|
||||
try:
|
||||
assert (
|
||||
db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none()
|
||||
== None
|
||||
is None
|
||||
)
|
||||
except AssertionError:
|
||||
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
import os
|
||||
import secrets
|
||||
import string
|
||||
@@ -15,7 +13,9 @@ from pwdlib import PasswordHash
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.model.models import Token, TokenData, UserDB
|
||||
from app.services.database import *
|
||||
from app.services.database import add_and_refresh, get_session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SECRET_KEY = os.getenv("SECRET_KEY", default="ff" * 16)
|
||||
ALGORITHM = "HS256"
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
import logging
|
||||
from os import getenv, path
|
||||
from sqlmodel import Session, SQLModel, create_engine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
from os import getenv, path
|
||||
|
||||
from sqlmodel import Session, SQLModel, create_engine
|
||||
|
||||
SQLALCHEMY_DATABASE_URL = getenv("SQLALCHEMY_DATABASE_URL", "sqlite:///./gatekeeper.db")
|
||||
|
||||
|
||||
@@ -1,14 +1,14 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
from datetime import date, datetime, timedelta
|
||||
|
||||
from sqlalchemy import exc
|
||||
from sqlmodel import select
|
||||
|
||||
from app.model.models import *
|
||||
from app.model.models import Card, OneShotAccess
|
||||
from app.services.database import Session, add_and_refresh
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
doorIsOpen = True
|
||||
# I think this could also be gpio controlled
|
||||
# See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
import os
|
||||
import secrets
|
||||
import threading
|
||||
@@ -30,6 +27,8 @@ from smartcard.Exceptions import CardRequestTimeoutException
|
||||
|
||||
from app.services.door import checkAccess, openDoor
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ENV vars
|
||||
load_dotenv()
|
||||
MIFARE_APP_MASTER_KEY = os.getenv("MIFARE_APP_MASTER_KEY")
|
||||
@@ -45,7 +44,7 @@ MIFARE_ENCRYPTED_FILE_ID = 0x1
|
||||
|
||||
|
||||
def checkForKey():
|
||||
if MIFARE_APP_MASTER_KEY == None:
|
||||
if MIFARE_APP_MASTER_KEY is None:
|
||||
logger.critical("NO MASTER KEY LOADED")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
@@ -334,7 +333,7 @@ class BackgroundScanner:
|
||||
|
||||
def _check_db(self, key):
|
||||
check = checkAccess(key, self.db)
|
||||
if check == True:
|
||||
if check:
|
||||
openDoor()
|
||||
logger.info("Access granted!")
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user