wow, thats pretty ruff

ran 'ruff check --fix app'
This commit is contained in:
2026-07-29 01:41:26 +02:00
parent ede3e9a516
commit 1ca046ded9
14 changed files with 120 additions and 92 deletions

View File

@@ -1,5 +1,6 @@
from fastapi import FastAPI
from .controllers import userManager, cardManager
from .controllers import cardManager, userManager
app = FastAPI()
app.include_router(userManager.user_router)

View File

@@ -1,21 +1,21 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from sqlalchemy.orm import selectinload
from typing import List
from sqlmodel import Session, select
from app.model.models import *
from app.services.database import engine, get_session, add_and_refresh
from app.services.auth import auth_is_admin
import uuid as gen_uuid
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)):
logger.info(f"Creating accessauth with data: {aa}")
if aa.timetables is not []:
if aa.timetables != []:
timetables = [Timetable.model_validate(t) for t in aa.timetables]
else: timetables = []
if aa.oneshot is not None:
@@ -30,7 +30,7 @@ def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizatio
)
return add_and_refresh(db, db_aa)
@aa_router.get("/", response_model=List[AccessAuthorizationResponse])
@aa_router.get("/", response_model=list[AccessAuthorizationResponse])
def get_all_accessauths(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
return db.exec(
select(AccessAuthorizationDB)

View File

@@ -1,15 +1,15 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List
from sqlalchemy.exc import NoResultFound
from ..model.models import Card, CardCreate, CardUpdate, GroupDB
from ..services.database import engine, get_session, add_and_refresh
from ..services.auth import auth_is_admin
import uuid as gen_uuid
from app.services.scanner import WriteNewCard, DeleteCard
logger = logging.getLogger(__name__)
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
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
@@ -51,7 +51,7 @@ def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_i
db.commit()
return {"message": "Card deleted successfully"}
@card_router.get("/{group_id}", response_model=List[Card])
@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

View File

@@ -1,11 +1,13 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException
from app.services.auth import auth_is_admin
from sqlalchemy import exc
from sqlmodel import Session, select
import sqlalchemy.exc as exc
from app.model.models import *
from app.services.database import get_session, add_and_refresh
from app.services.auth import auth_is_admin
from app.services.database import add_and_refresh, get_session
debug_router = APIRouter(
prefix="/api/v1/debug",
@@ -41,7 +43,7 @@ def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
@debug_router.put("/getcards")
def list_all_cards(db: Session = Depends(get_session)):
logger.info(f"Debug Setting: Getting cards.")
logger.info("Debug Setting: Getting cards.")
cards = db.exec(select(Card)).all()
print(cards)
out = []

View File

@@ -1,9 +1,9 @@
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from sqlmodel import Session
from app.services.database import get_session
from app.services.auth import auth_is_admin
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"])

View File

@@ -1,14 +1,14 @@
from fastapi import APIRouter, HTTPException, Depends, status
from sqlmodel import Session, select
from typing import List
from ..model.models import GroupDB, GroupResponse, GroupCreate
from ..services.database import engine, get_session, add_and_refresh
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from ..model.models import GroupCreate, GroupDB, GroupResponse
from ..services.auth import auth_is_admin
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])
@group_router.get("/", response_model=list[GroupResponse])
def get_groups(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
groups = db.exec(select(GroupDB)).all()
return groups

View File

@@ -1,12 +1,14 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, HTTPException, Depends, status
from sqlmodel import Session, select
from typing import List
from ..model.models import UserResponse, UserCreate, UserDB, UserUpdate
from ..services.database import engine, get_session, add_and_refresh
from ..services.auth import get_password_hash, get_current_user as auth_user, auth_is_admin
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
user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
@@ -19,7 +21,7 @@ def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin:
db_user = UserDB.model_validate(user, update=hashed_password)
return add_and_refresh(db, db_user)
@user_router.get("/", response_model=List[UserResponse])
@user_router.get("/", response_model=list[UserResponse])
def read_users(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
users = db.exec(select(UserDB)).all()
return users

View File

@@ -1,19 +1,28 @@
import logging
logger = logging.getLogger(__name__)
import os
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer
from contextlib import asynccontextmanager
from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer
load_dotenv()
from app.controllers import userManager, cardManager, groupManager, aaManager, doorManager, debugManager
from app.controllers import (
aaManager,
cardManager,
debugManager,
doorManager,
groupManager,
userManager,
)
from app.services.auth import create_first_user, token_router
from app.services.database import create_db_and_tables, get_db_session
from app.services.auth import token_router, create_first_user
from app.services.settings import verify_settings
from app.services.scanner import BackgroundScanner
from app.services.settings import verify_settings
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
scanner = BackgroundScanner(db=get_db_session())

View File

@@ -1,7 +1,9 @@
from sqlmodel import Field, Relationship, Session, SQLModel
from typing import List, Literal, Union
from datetime import datetime, time
from typing import Literal
from pydantic import model_validator
from sqlmodel import Field, Relationship, SQLModel
class Base(SQLModel):
pass
@@ -50,13 +52,13 @@ class GroupCreate(GroupBase):
class GroupDB(GroupBase, table=True):
id: int | None = Field(default=None, primary_key=True)
cards: List["Card"] = Relationship(back_populates="group")
accessauths: List["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink)
cards: list["Card"] = Relationship(back_populates="group")
accessauths: list["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink)
class GroupResponse(GroupBase):
id: int
cards: List["Card"] | None
accessauths: List["AccessAuthorizationDB"] | None
cards: list["Card"] | None
accessauths: list["AccessAuthorizationDB"] | None
#### AccessAuthorization
class AccessAuthorizationBase(Base):
@@ -67,8 +69,8 @@ class AccessAuthorizationBase(Base):
class AccessAuthorizationDB(AccessAuthorizationBase, table=True):
id: int | None = Field(default=None, primary_key=True)
type: str
groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
timetables: List["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True)
groups: list["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
timetables: list["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True)
oneshot: "OneShotAccess" = Relationship(back_populates="accessauth", cascade_delete=True)
class OneShotAccessBase(Base):
@@ -81,7 +83,7 @@ class OneShotAccess(OneShotAccessBase, table=True):
accessauth: AccessAuthorizationDB = Relationship(back_populates="oneshot")
class AccessAuthorizationCreate(AccessAuthorizationBase):
timetables: List["TimetableCreate"] = []
timetables: list["TimetableCreate"] = []
oneshot: OneShotAccessBase | None = None
@model_validator(mode="after")
@@ -102,15 +104,15 @@ class AccessAuthorizationCreate(AccessAuthorizationBase):
class AccessAuthorizationResponse(AccessAuthorizationBase):
id: int
timetables: List["Timetable"] = []
timetables: list["Timetable"] = []
oneshot: OneShotAccessBase | None = None
groups: List["GroupDB"]
groups: list["GroupDB"]
class AccessAuthorizationUpdate(Base):
name: str | None = None
type: Literal["timetable", "oneshot", "somefuturespec"] | None = None
is_active: bool | None = None
timetables: List["TimetableCreate"] | None = None
timetables: list["TimetableCreate"] | None = None
oneshot: OneShotAccessBase | None = None

View File

@@ -1,16 +1,21 @@
import logging
logger = logging.getLogger(__name__)
import os
import secrets
import string
from datetime import UTC, datetime, timedelta
from typing import Annotated
from datetime import datetime, timedelta, timezone
from fastapi import APIRouter, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlmodel import Session, select
from pwdlib import PasswordHash
import jwt
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from app.model.models import UserDB, Token, TokenData, UserCreate
from pwdlib import PasswordHash
from sqlmodel import Session, select
from app.model.models import Token, TokenData, UserDB
from app.services.database import *
import secrets, string, os
SECRET_KEY = os.getenv("SECRET_KEY", default="ff"*16)
ALGORITHM = "HS256"
@@ -43,9 +48,9 @@ def authenticate_user(db, username: str, password: str):
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
expire = datetime.now(UTC) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
expire = datetime.now(UTC) + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt

View File

@@ -1,9 +1,9 @@
import logging
logger = logging.getLogger(__name__)
from os import getenv, path
from sqlmodel import create_engine, SQLModel, Session
from app.model.models import Base
from sqlmodel import Session, SQLModel, create_engine
SQLALCHEMY_DATABASE_URL = getenv("SQLALCHEMY_DATABASE_URL", "sqlite:///./gatekeeper.db")
@@ -13,6 +13,7 @@ def create_db_and_tables():
if not path.exists(SQLALCHEMY_DATABASE_URL):
SQLModel.metadata.create_all(engine)
from alembic.config import Config
from alembic import command
alembic_cfg = Config("./alembic.ini")
command.stamp(alembic_cfg, "head")

View File

@@ -1,13 +1,13 @@
import logging
logger = logging.getLogger(__name__)
from sqlmodel import select
from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import selectinload
import sqlalchemy.exc as exc
from datetime import datetime, date, timedelta
from app.services.database import Session, get_session, add_and_refresh
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.services.database import Session, add_and_refresh
doorIsOpen = True
# I think this could also be gpio controlled
@@ -17,13 +17,11 @@ def openDoor():
global doorIsOpen
doorIsOpen = True
logger.info("Still needs gpio out")
pass
def closeDoor():
global doorIsOpen
doorIsOpen = False
logger.info("Still needs gpio out")
pass
def isDoorOpen():
return doorIsOpen

View File

@@ -1,27 +1,34 @@
import logging
logger = logging.getLogger(__name__)
import threading
import time
import os
import secrets
import threading
import time
from typing import Optional
from sqlmodel import Session
from desfire import (
DESFire,
DESFireKey,
PCSCDevice,
diversify_key,
get_list,
to_hex_string,
)
from desfire.enums import (
DESFireCommunicationMode,
DESFireFileType,
DESFireKeySettings,
DESFireKeyType,
)
from desfire.schemas import FilePermissions, FileSettings, KeySettings
from dotenv import load_dotenv
from fastapi import HTTPException, status
from smartcard.CardRequest import CardRequest
from smartcard.CardType import AnyCardType
from smartcard.Exceptions import CardRequestTimeoutException
from desfire import DESFire, DESFireKey, PCSCDevice, diversify_key, get_list, to_hex_string
from desfire.enums import DESFireCommunicationMode, DESFireFileType, DESFireKeySettings, DESFireKeyType
from desfire.schemas import FilePermissions, FileSettings, KeySettings
import desfire.exceptions as desExceptions
from app.services.door import openDoor, closeDoor, isDoorOpen, checkAccess
from app.services.door import checkAccess, openDoor
#ENV vars
load_dotenv()
@@ -112,7 +119,7 @@ def DeleteCard():
try:
try:
logger.debug("Auth1")#
logger.debug("Auth1")
desfire.authenticate(0x0, aes_master_key)
except:
logger.debug("Auth2")
@@ -244,7 +251,7 @@ class BackgroundScanner:
def __init__(self, db):
self.db = db
self.is_running = False
self.thread: Optional[threading.Thread] = None
self.thread: threading.Thread | None = None
def start(self):
if self.is_running:

View File

@@ -1,5 +1,6 @@
import os
import logging
import os
logger = logging.getLogger(__name__)
def verify_settings():