Backend Refactoring of CRUD and Models

backend/app/alembic/env.py

@@ -18,9 +18,10 @@
 # target_metadata = mymodel.Base.metadata
 # target_metadata = None
 
-from app.models import SQLModel  # noqa
+from sqlmodel import SQLModel  # Direkter Import aus dem sqlmodel Paket
 from app.core.config import settings # noqa
 
+
 target_metadata = SQLModel.metadata
 
 # other values from the config, defined by the needs of env.py,

backend/app/api/deps.py

@@ -11,7 +11,7 @@
 from app.core import security
 from app.core.config import settings
 from app.core.db import engine
-from app.models import TokenPayload, User
+from app.models.user import User, TokenPayload
 
 reusable_oauth2 = OAuth2PasswordBearer(
     tokenUrl=f"{settings.API_V1_STR}/login/access-token"

backend/app/api/routes/items.py

@@ -5,7 +5,8 @@
 from sqlmodel import func, select
 
 from app.api.deps import CurrentUser, SessionDep
-from app.models import Item, ItemCreate, ItemPublic, ItemsPublic, ItemUpdate, Message
+from app.models.user import Message
+from app.models.item import Item, ItemCreate, ItemPublic, ItemsPublic, ItemUpdate
 
 router = APIRouter()
 

backend/app/api/routes/login.py

@@ -5,12 +5,12 @@
 from fastapi.responses import HTMLResponse
 from fastapi.security import OAuth2PasswordRequestForm
 
-from app import crud
+from app.crud import authenticate, get_user_by_email
 from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser
 from app.core import security
 from app.core.config import settings
 from app.core.security import get_password_hash
-from app.models import Message, NewPassword, Token, UserPublic
+from app.models.user import Message, NewPassword, Token, UserPublic
 from app.utils import (
     generate_password_reset_token,
     generate_reset_password_email,
@@ -28,7 +28,7 @@ def login_access_token(
     """
     OAuth2 compatible token login, get an access token for future requests
     """
-    user = crud.authenticate(
+    user = authenticate(
         session=session, email=form_data.username, password=form_data.password
     )
     if not user:
@@ -56,7 +56,7 @@ def recover_password(email: str, session: SessionDep) -> Message:
     """
     Password Recovery
     """
-    user = crud.get_user_by_email(session=session, email=email)
+    user = get_user_by_email(session=session, email=email)
 
     if not user:
         raise HTTPException(
@@ -83,7 +83,7 @@ def reset_password(session: SessionDep, body: NewPassword) -> Message:
     email = verify_password_reset_token(token=body.token)
     if not email:
         raise HTTPException(status_code=400, detail="Invalid token")
-    user = crud.get_user_by_email(session=session, email=email)
+    user = get_user_by_email(session=session, email=email)
     if not user:
         raise HTTPException(
             status_code=404,
@@ -107,7 +107,7 @@ def recover_password_html_content(email: str, session: SessionDep) -> Any:
     """
     HTML Content for Password Recovery
     """
-    user = crud.get_user_by_email(session=session, email=email)
+    user = get_user_by_email(session=session, email=email)
 
     if not user:
         raise HTTPException(

backend/app/api/routes/users.py

@@ -4,7 +4,8 @@
 from fastapi import APIRouter, Depends, HTTPException
 from sqlmodel import col, delete, func, select
 
-from app import crud
+from app.crud import get_user_by_email, create_user, update_user
+
 from app.api.deps import (
     CurrentUser,
     SessionDep,

backend/app/api/routes/utils.py

@@ -2,7 +2,7 @@
 from pydantic.networks import EmailStr
 
 from app.api.deps import get_current_active_superuser
-from app.models import Message
+from app.models.user import Message
 from app.utils import generate_test_email, send_email
 
 router = APIRouter()

backend/app/core/db.py

@@ -1,6 +1,6 @@
 from sqlmodel import Session, create_engine, select
 
-from app import crud
+from app.crud import create_user  # Direkt importiert
 from app.core.config import settings
 from app.models import User, UserCreate
 
@@ -30,4 +30,4 @@ def init_db(session: Session) -> None:
             password=settings.FIRST_SUPERUSER_PASSWORD,
             is_superuser=True,
         )
-        user = crud.create_user(session=session, user_create=user_in)
+        user = create_user(session=session, user_create=user_in)  # Direkter Aufruf von create_user

backend/app/crud/__init__.py

@@ -0,0 +1,12 @@
+from .crud_user import (
+    create_user,
+    update_user,
+    get_user_by_email,
+    authenticate,
+)
+
+from .crud_item import (
+    create_item,
+    get_item,
+    get_items,
+)

backend/app/crud/crud_item.py

@@ -0,0 +1,16 @@
+from sqlmodel import Session, select
+from app.models.item import Item, ItemCreate
+
+def create_item(*, session: Session, item_create: ItemCreate) -> Item:
+    db_item = Item(**item_create.dict())
+    session.add(db_item)
+    session.commit()
+    session.refresh(db_item)
+    return db_item
+
+def get_item(*, session: Session, item_id: int) -> Item | None:
+    return session.get(Item, item_id)
+
+def get_items(*, session: Session, skip: int = 0, limit: int = 10):
+    statement = select(Item).offset(skip).limit(limit)
+    return session.exec(statement).all()

backend/app/crud.py → backend/app/crud/crud_user.py

@@ -1,10 +1,7 @@
-import uuid
-from typing import Any
-
 from sqlmodel import Session, select
-
 from app.core.security import get_password_hash, verify_password
-from app.models import Item, ItemCreate, User, UserCreate, UserUpdate
+from app.models.user import User, UserCreate, UserUpdate
+from typing import Any
 
 
 def create_user(*, session: Session, user_create: UserCreate) -> User:
@@ -16,7 +13,6 @@ def create_user(*, session: Session, user_create: UserCreate) -> User:
     session.refresh(db_obj)
     return db_obj
 
-
 def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any:
     user_data = user_in.model_dump(exclude_unset=True)
     extra_data = {}
@@ -30,25 +26,15 @@ def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any:
     session.refresh(db_user)
     return db_user
 
-
 def get_user_by_email(*, session: Session, email: str) -> User | None:
     statement = select(User).where(User.email == email)
     session_user = session.exec(statement).first()
     return session_user
 
-
 def authenticate(*, session: Session, email: str, password: str) -> User | None:
     db_user = get_user_by_email(session=session, email=email)
     if not db_user:
         return None
     if not verify_password(password, db_user.hashed_password):
         return None
     return db_user
-
-
-def create_item(*, session: Session, item_in: ItemCreate, owner_id: uuid.UUID) -> Item:
-    db_item = Item.model_validate(item_in, update={"owner_id": owner_id})
-    session.add(db_item)
-    session.commit()
-    session.refresh(db_item)
-    return db_item

backend/app/models/__init__.py

@@ -0,0 +1,13 @@
+from .user import (
+    User,
+    UserCreate,
+    UserUpdate,
+    Message,
+    UpdatePassword,
+    UserPublic,
+    UserRegister,
+    UsersPublic,
+    UserUpdateMe
+)
+
+from .item import Item

backend/app/models/item.py

@@ -0,0 +1,44 @@
+
+import uuid
+from sqlmodel import Field, SQLModel, Relationship
+from typing import Optional
+from datetime import date
+
+# Shared properties
+class ItemBase(SQLModel):
+    title: str = Field(min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=255)
+    aiwscode: str = Field(min_length=1, max_length=50)
+    name: str = Field(min_length=1, max_length=255)
+    location: str = Field(min_length=1, max_length=255)
+    expiry: Optional[date] = Field(default=None)
+    stk: Optional[date] = Field(default=None)
+    mtk: Optional[date] = Field(default=None)
+    lot: Optional[str] = Field(default=None, max_length=255)
+    serial: Optional[str] = Field(default=None, max_length=255)
+    notes: Optional[str] = Field(default=None, max_length=1000)
+
+# Properties to receive on item creation
+class ItemCreate(ItemBase):
+    pass
+
+# Properties to receive on item update
+class ItemUpdate(ItemBase):
+    pass
+
+# Database model, database table inferred from class name
+class Item(ItemBase, table=True):
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    owner_id: uuid.UUID = Field(
+        foreign_key="user.id", nullable=False, ondelete="CASCADE"
+    )
+    owner: Optional['User'] = Relationship(back_populates="items")  # Lazy import for User using a string
+
+# Properties to return via API, id is always required
+class ItemPublic(ItemBase):
+    id: uuid.UUID
+    owner_id: uuid.UUID
+
+class ItemsPublic(SQLModel):
+    data: list[ItemPublic]
+    count: int