switch orm to sqlmodel

app/controllers/cardManager.py

@@ -1,56 +1,57 @@
 from fastapi import APIRouter, Depends, HTTPException
-from sqlalchemy.orm import Session
+from sqlmodel import Session
-from sqlalchemy import select
 from typing import List
 
 from ..model.dbModels import Card, AccessAuthorization
 from ..model.card import CardBase, AccessAuthorizationCreate, AccessAuthorization as AccessSchema
-from ..services.database import SessionLocal, engine
+from ..services.database import engine
 import uuid as gen_uuid
 
 card_router = APIRouter(tags=["Card"])
 
-def get_db():
+def register_card(name: str):
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
-def register_card(db: Session, name: str):
     uuid = str(gen_uuid.uuid4()) #hier code für mifare registrierung
     card = Card(user_id=name, uuid=uuid)
     return card
 
 @card_router.get("/cards", response_model=List[AccessSchema])
-def get_accesses(db: Session = Depends(get_db)):
+def get_accesses():
-    accesses = db.query(AccessAuthorization).all()
+    with Session(engine) as db:
-    if accesses is None:
+        accesses = db.query(AccessAuthorization).all()
-        raise HTTPException(status_code=404, detail="N/A")
+        if accesses is None:
-    return accesses
+            raise HTTPException(status_code=404, detail="N/A")
+        return accesses
 
 @card_router.post("/cards", response_model=AccessSchema)
-def create_access(access: AccessAuthorizationCreate, db: Session = Depends(get_db)):
+def create_access(access: AccessAuthorizationCreate):
-    db_access = AccessAuthorization(**access.dict())
+    with Session(engine) as db:
-    card = register_card(db, access.name)
+        db_access = AccessAuthorization(**access.dict())
-    db.add(db_access)
+        card = register_card(access.name)
-    db.add(card)
+        db.add(db_access)
-    db.commit()
+        db.add(card)
-    db.refresh(db_access)
+        db.commit()
-    return db_access
+        db.refresh(db_access)
+        return db_access
 
 @card_router.get("/cards/{auth_name}", response_model=List[CardBase])
-def get_cards(auth_name: str, db: Session = Depends(get_db)):
+def get_cards(auth_name: str):
-    stmt = select(AccessAuthorization).where(AccessAuthorization.name == auth_name)
+    with Session(engine) as db:
-    access_auth = db.execute(stmt).scalar_one_or_none()
+        stmt = select(AccessAuthorization).where(AccessAuthorization.name == auth_name)
-    if access_auth is None:
+        access_auth = db.execute(stmt).scalar_one_or_none()
-        raise HTTPException(status_code=404, detail="Not found!")
+        if access_auth is None:
-    return access_auth.card_id
+            raise HTTPException(status_code=404, detail="Not found!")
+        return access_auth.card_id
 
 @card_router.post("/cards/{auth_name}", response_model=CardBase)
-def add_card(auth_name: str, db: Session = Depends(get_db)):
+def add_card(auth_name: str):
-    card = register_card(db, auth_name)
+    with Session(engine) as db:
-    db.add(card)
+        card = register_card(auth_name)
-    db.commit()
+        db.add(card)
-    db.refresh(card)
+        db.commit()
-    return card
+        db.refresh(card)
+        return card
+
+#TODO:
+# -Split Authorisations + Cards
+# -Deactivation
+# -Deleting

app/controllers/userManager.py

@@ -1,56 +1,54 @@
 from fastapi import APIRouter, Depends, HTTPException
-from sqlalchemy.orm import Session
+from sqlmodel import Session
 from typing import List
 
 from ..model.dbModels import User
 from ..model.user import UserCreate, User as UserSchema
-from ..services.database import SessionLocal, engine
+from ..services.database import engine
 
 user_router = APIRouter(tags=["users"])
 
-def get_db():
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
 @user_router.post("/users/", response_model=UserSchema)
-def create_user(user: UserCreate, db: Session = Depends(get_db)):
+def create_user(user: UserCreate):
-    db_user = User(**user.dict())
+    with Session(engine) as db:
-    db.add(db_user)
+        db_user = User(**user.dict())
-    db.commit()
+        db.add(db_user)
-    db.refresh(db_user)
+        db.commit()
-    return db_user
+        db.refresh(db_user)
+        return db_user
 
 @user_router.get("/users/", response_model=List[UserSchema])
-def read_users(db: Session = Depends(get_db)):
+def read_users():
-    users = db.query(User).all()
+    with Session(engine) as db:
-    return users
+        users = db.query(User).all()
+        return users
 
 @user_router.get("/users/{user_id}", response_model=UserSchema)
-def read_user(user_id: int, db: Session = Depends(get_db)):
+def read_user(user_id: int):
-    db_user = db.query(User).filter(User.id == user_id).first()
+    with Session(engine) as db:
-    if db_user is None:
+        db_user = db.query(User).filter(User.id == user_id).first()
-        raise HTTPException(status_code=404, detail="User not found")
+        if db_user is None:
-    return db_user
+            raise HTTPException(status_code=404, detail="User not found")
+        return db_user
 
 @user_router.put("/users/{user_id}", response_model=UserSchema)
-def update_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
+def update_user(user_id: int, user: UserCreate):
-    db_user = db.query(User).filter(User.id == user_id).first()
+    with Session(engine) as db:
-    if db_user is None:
+        db_user = db.query(User).filter(User.id == user_id).first()
-        raise HTTPException(status_code=404, detail="User not found")
+        if db_user is None:
-    for key, value in user.dict().items():
+            raise HTTPException(status_code=404, detail="User not found")
-        setattr(db_user, key, value)
+        for key, value in user.dict().items():
-    db.commit()
+            setattr(db_user, key, value)
-    db.refresh(db_user)
+        db.commit()
-    return db_user
+        db.refresh(db_user)
+        return db_user
 
 @user_router.delete("/users/{user_id}")
-def delete_user(user_id: int, db: Session = Depends(get_db)):
+def delete_user(user_id: int):
-    db_user = db.query(User).filter(User.id == user_id).first()
+    with Session(engine) as db:
-    if db_user is None:
+        db_user = db.query(User).filter(User.id == user_id).first()
-        raise HTTPException(status_code=404, detail="User not found")
+        if db_user is None:
-    db.delete(db_user)
+            raise HTTPException(status_code=404, detail="User not found")
-    db.commit()
+        db.delete(db_user)
-    return {"message": "User deleted successfully"}
+        db.commit()
+        return {"message": "User deleted successfully"}

app/model/dbModels.py

@@ -1,35 +1,43 @@
-from __future__ import annotations
+from sqlmodel import Field, Relationship, Session, SQLModel
-from typing import List, Optional
+from typing import List
 
-from sqlalchemy import ForeignKey
+class Base(SQLModel):
-from sqlalchemy import Column, Integer, String, Boolean
-from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase, relationship
-
-class Base(DeclarativeBase):
     pass
 
-class User(Base):
+class User(Base, table=True):
-    __tablename__ = "users"
+    id: int | None = Field(default=None, primary_key=True)
+    name: str = Field(index=True)
+    email: str | None = None
+    password: str
+    is_admin: bool
 
-    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+class AaGroupLink(Base, table=True):
-    name: Mapped[str]
+    group_id: int | None = Field(default=None, foreign_key="group.id", primary_key=True)
-    email: Mapped[Optional[str]]
+    accessauth_id: int | None = Field(default=None, foreign_key="accessauthorization.id", primary_key=True)
-    password: Mapped[str]
-    is_admin: Mapped[bool]
 
-class AccessAuthorization(Base): #parent
+class Group(Base, table=True):
-    __tablename__ = "access_authorizations"
+    id: int | None = Field(default=None, primary_key=True)
+    name: str = Field(primary_key=True)
+    cards: List["Card"] = Relationship(back_populates="group")
+    accessauths: List["AccessAuthorization"] = Relationship(back_populates="groups", link_model=AaGroupLink)
 
-    name: Mapped[str] = mapped_column(primary_key=True, index=True)
+class AccessAuthorization(Base, table=True):
-    is_active: Mapped[bool]
+    id: int | None = Field(default=None, primary_key=True)
-    card_id: Mapped[List["Card"]] = relationship(back_populates="user")
+    name: str = Field(index=True)
-
+    is_active: bool
-class Card(Base): #child
+    groups: list["Group"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
-    __tablename__ = "cards"
+    timetables: list["Timetable"] = Relationship(back_populates="accessauth")
-
-    id: Mapped[int] = mapped_column(primary_key=True, index=True)
-    uuid: Mapped[str] = mapped_column(index=True)
-    user_id = mapped_column(ForeignKey("access_authorizations.name"))
-    user: Mapped["AccessAuthorization"] = relationship(back_populates="card_id")
 
+class Card(Base, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    uuid: str
+    group_id: int | None = Field(default=None, foreign_key="group.id")
+    group: Group | None = Relationship(back_populates="cards")
 
+class Timetable(Base, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    weekday: int
+    starttime: str
+    duration: int
+    accessauth_id: int | None = Field(default=None, foreign_key="accessauthorization.id")
+    accessauth: AccessAuthorization | None = Relationship(back_populates="timetables")

app/services/database.py

@@ -1,13 +1,8 @@
-from sqlalchemy import create_engine
+from sqlmodel import create_engine, SQLModel
-from sqlalchemy.orm import sessionmaker
 
 from ..model.dbModels import Base
 
 SQLALCHEMY_DATABASE_URL = "sqlite:///./gatekeeper.db"
 
-engine = create_engine(
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
-    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
+SQLModel.metadata.create_all(engine)
-)
-SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-Base.metadata.create_all(bind=engine)

pyproject.toml

@@ -6,9 +6,9 @@ readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "fastapi[standard]>=0.135.3",
+    "sqlmodel>=0.0.38",
     "poetry>=2.3.4",
     "python-desfire",
-    "sqlalchemy>=2.0.49",
 ]
 
 [tool.uv.sources]

uv.lock

@@ -481,7 +481,7 @@ dependencies = [
     { name = "fastapi", extra = ["standard"] },
     { name = "poetry" },
     { name = "python-desfire" },
-    { name = "sqlalchemy" },
+    { name = "sqlmodel" },
 ]
 
 [package.metadata]
@@ -489,7 +489,7 @@ requires-dist = [
     { name = "fastapi", extras = ["standard"], specifier = ">=0.135.3" },
     { name = "poetry", specifier = ">=2.3.4" },
     { name = "python-desfire", git = "https://github.com/waza-ari/python-desfire" },
-    { name = "sqlalchemy", specifier = ">=2.0.49" },
+    { name = "sqlmodel", specifier = ">=0.0.38" },
 ]
 
 [[package]]
@@ -1350,6 +1350,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e5/30/8519fdde58a7bdf155b714359791ad1dc018b47d60269d5d160d311fdc36/sqlalchemy-2.0.49-py3-none-any.whl", hash = "sha256:ec44cfa7ef1a728e88ad41674de50f6db8cfdb3e2af84af86e0041aaf02d43d0", size = 1942158, upload-time = "2026-04-03T16:53:44.135Z" },
 ]
 
+[[package]]
+name = "sqlmodel"
+version = "0.0.38"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pydantic" },
+    { name = "sqlalchemy" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/64/0d/26ec1329960ea9430131fe63f63a95ea4cb8971d49c891ff7e1f3255421c/sqlmodel-0.0.38.tar.gz", hash = "sha256:d583ec237b14103809f74e8630032bc40ab68cd6b754a610f0813c56911a547b", size = 86710, upload-time = "2026-04-02T21:03:55.571Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/72/c7/10c60af0607ab6fa136264f7f39d205932218516226d38585324ffda705d/sqlmodel-0.0.38-py3-none-any.whl", hash = "sha256:84e3fa990a77395461ded72a6c73173438ce8449d5c1c4d97fbff1b1df692649", size = 27294, upload-time = "2026-04-02T21:03:56.406Z" },
+]
+
 [[package]]
 name = "starlette"
 version = "1.0.0"