Finish sqlmodel migration

- merge card.py, user.py and dbmodels.py into models.py
- deduplicate
- change responsemodels in the manager functions
- correct inports
This commit is contained in:
2026-04-23 22:41:48 +02:00
parent ba0f036535
commit 22e57996c2
7 changed files with 83 additions and 87 deletions

View File

@@ -2,8 +2,7 @@ from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from typing import List
from ..model.dbModels import Card, AccessAuthorization
from ..model.card import CardBase, AccessAuthorizationCreate, AccessAuthorization as AccessSchema
from ..model.models import Card, AccessAuthorizationDB, AccessAuthorizationCreate, AccessAuthorizationResponse
from ..services.database import engine
import uuid as gen_uuid
@@ -14,7 +13,7 @@ def register_card(name: str):
card = Card(user_id=name, uuid=uuid)
return card
@card_router.get("/cards", response_model=List[AccessSchema])
@card_router.get("/cards", response_model=List[AccessAuthorizationResponse])
def get_accesses():
with Session(engine) as db:
accesses = db.query(AccessAuthorization).all()
@@ -22,7 +21,7 @@ def get_accesses():
raise HTTPException(status_code=404, detail="N/A")
return accesses
@card_router.post("/cards", response_model=AccessSchema)
@card_router.post("/cards", response_model=AccessAuthorizationResponse)
def create_access(access: AccessAuthorizationCreate):
with Session(engine) as db:
db_access = AccessAuthorization(**access.dict())
@@ -33,7 +32,7 @@ def create_access(access: AccessAuthorizationCreate):
db.refresh(db_access)
return db_access
@card_router.get("/cards/{auth_name}", response_model=List[CardBase])
@card_router.get("/cards/{auth_name}", response_model=List[Card])
def get_cards(auth_name: str):
with Session(engine) as db:
stmt = select(AccessAuthorization).where(AccessAuthorization.name == auth_name)
@@ -42,7 +41,7 @@ def get_cards(auth_name: str):
raise HTTPException(status_code=404, detail="Not found!")
return access_auth.card_id
@card_router.post("/cards/{auth_name}", response_model=CardBase)
@card_router.post("/cards/{auth_name}", response_model=Card)
def add_card(auth_name: str):
with Session(engine) as db:
card = register_card(auth_name)

View File

@@ -2,13 +2,12 @@ from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from typing import List
from ..model.dbModels import User
from ..model.user import UserCreate, User as UserSchema
from ..model.models import UserResponse, UserCreate, UserDB
from ..services.database import engine
user_router = APIRouter(tags=["users"])
@user_router.post("/users/", response_model=UserSchema)
@user_router.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate):
with Session(engine) as db:
db_user = User(**user.dict())
@@ -17,13 +16,13 @@ def create_user(user: UserCreate):
db.refresh(db_user)
return db_user
@user_router.get("/users/", response_model=List[UserSchema])
@user_router.get("/users/", response_model=List[UserResponse])
def read_users():
with Session(engine) as db:
users = db.query(User).all()
return users
@user_router.get("/users/{user_id}", response_model=UserSchema)
@user_router.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int):
with Session(engine) as db:
db_user = db.query(User).filter(User.id == user_id).first()
@@ -31,7 +30,7 @@ def read_user(user_id: int):
raise HTTPException(status_code=404, detail="User not found")
return db_user
@user_router.put("/users/{user_id}", response_model=UserSchema)
@user_router.put("/users/{user_id}", response_model=UserResponse)
def update_user(user_id: int, user: UserCreate):
with Session(engine) as db:
db_user = db.query(User).filter(User.id == user_id).first()

View File

@@ -1,17 +0,0 @@
from pydantic import BaseModel
class CardBase(BaseModel):
id: int
uuid: str
user_id: str
class AccessAuthorizationBase(BaseModel):
name: str
is_active: bool = True
class AccessAuthorizationCreate(AccessAuthorizationBase):
pass
class AccessAuthorization(AccessAuthorizationBase):
class Config:
from_attributes = True

View File

@@ -1,43 +0,0 @@
from sqlmodel import Field, Relationship, Session, SQLModel
from typing import List
class Base(SQLModel):
pass
class User(Base, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
email: str | None = None
password: str
is_admin: bool
class AaGroupLink(Base, table=True):
group_id: int | None = Field(default=None, foreign_key="group.id", primary_key=True)
accessauth_id: int | None = Field(default=None, foreign_key="accessauthorization.id", primary_key=True)
class Group(Base, table=True):
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)
class AccessAuthorization(Base, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
is_active: bool
groups: list["Group"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
timetables: list["Timetable"] = Relationship(back_populates="accessauth")
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")

72
app/model/models.py Normal file
View File

@@ -0,0 +1,72 @@
from sqlmodel import Field, Relationship, Session, SQLModel
from typing import List
class Base(SQLModel):
pass
#### User
class UserBase(Base):
name: str = Field(index=True)
email: str | None = None
is_admin: bool = False
class UserResponse(UserBase):
id: int
class UserCreate(UserBase):
password: str
class UserDB(UserBase, table=True):
id: int | None = Field(default=None, primary_key=True)
password: str
#### Special
class AaGroupLink(Base, table=True):
group_id: int | None = Field(default=None, foreign_key="group.id", primary_key=True)
accessauth_id: int | None = Field(default=None, foreign_key="accessauthorization.id", primary_key=True)
#### Group
class GroupBase(Base):
name: str = Field(index=True)
class GroupDB(GroupBase, table=True):
id: int | None = Field(default=None, primary_key=True)
cards: List["Card"] = Relationship(back_populates="group")
accessauths: List["AccessAuthorizationBase"] = Relationship(back_populates="groups", link_model=AaGroupLink)
class GroupResponse(GroupBase):
id: int
cards: List["Card"]
accessauths: List["AccessAuthorizationBase"]
#### AccessAuthorization
class AccessAuthorizationBase(Base):
name: str = Field(index=True)
is_active: bool
groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
timetables: List["Timetable"] = Relationship(back_populates="accessauth")
class AccessAuthorizationDB(Base, table=True):
id: int | None = Field(default=None, primary_key=True)
class AccessAuthorizationCreate(AccessAuthorizationBase):
pass
class AccessAuthorizationResponse(AccessAuthorizationBase):
id: int
#### Card
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="groupdb.id")
group: GroupDB | 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="accessauthorizationdb.id")
accessauth: AccessAuthorizationDB | None = Relationship(back_populates="timetables")

View File

@@ -1,14 +0,0 @@
from pydantic import BaseModel
class UserBase(BaseModel):
name: str
email: str | None = None
is_admin: bool = False
class UserCreate(UserBase):
password: str
class User(UserBase):
id: int
class Config:
from_attributes = True

View File

@@ -1,6 +1,6 @@
from sqlmodel import create_engine, SQLModel
from ..model.dbModels import Base
from ..model.models import Base
SQLALCHEMY_DATABASE_URL = "sqlite:///./gatekeeper.db"