Replace SQLModel with SQLAlchemy in items routes

backend/app/alembic/env.py

@@ -18,10 +18,10 @@
 # target_metadata = mymodel.Base.metadata
 # target_metadata = None
 
-from app.models import SQLModel  # noqa
+from app.models import Base  # noqa
 from app.core.config import settings # noqa
 
-target_metadata = SQLModel.metadata
+target_metadata = Base.metadata
 
 # other values from the config, defined by the needs of env.py,
 # can be acquired:

backend/app/alembic/versions/1a31ce608336_add_cascade_delete_relationships.py

@@ -7,7 +7,7 @@
 """
 from alembic import op
 import sqlalchemy as sa
-import sqlmodel.sql.sqltypes
+from sqlalchemy.dialects import postgresql
 
 
 # revision identifiers, used by Alembic.

backend/app/alembic/versions/9c0a54914c78_add_max_length_for_string_varchar_.py

@@ -7,7 +7,7 @@
 """
 from alembic import op
 import sqlalchemy as sa
-import sqlmodel.sql.sqltypes
+from sqlalchemy.dialects import postgresql
 
 
 # revision identifiers, used by Alembic.

backend/app/alembic/versions/d98dd8ec85a3_edit_replace_id_integers_in_all_models_.py

@@ -7,7 +7,6 @@
 """
 from alembic import op
 import sqlalchemy as sa
-import sqlmodel.sql.sqltypes
 from sqlalchemy.dialects import postgresql
 
 

backend/app/alembic/versions/e2412789c190_initialize_models.py

@@ -6,7 +6,7 @@
 
 """
 import sqlalchemy as sa
-import sqlmodel.sql.sqltypes
+from sqlalchemy.dialects import postgresql
 from alembic import op
 
 # revision identifiers, used by Alembic.
@@ -20,22 +20,22 @@ def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
     op.create_table(
         "user",
-        sa.Column("email", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("email", sa.String(255), nullable=False),
         sa.Column("is_active", sa.Boolean(), nullable=False),
         sa.Column("is_superuser", sa.Boolean(), nullable=False),
-        sa.Column("full_name", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("full_name", sa.String(255), nullable=True),
         sa.Column("id", sa.Integer(), nullable=False),
         sa.Column(
-            "hashed_password", sqlmodel.sql.sqltypes.AutoString(), nullable=False
+            "hashed_password", sa.String(), nullable=False
         ),
         sa.PrimaryKeyConstraint("id"),
     )
     op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
     op.create_table(
         "item",
-        sa.Column("description", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
         sa.Column("id", sa.Integer(), nullable=False),
-        sa.Column("title", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("title", sa.String(255), nullable=False),
         sa.Column("owner_id", sa.Integer(), nullable=False),
         sa.ForeignKeyConstraint(
             ["owner_id"],

backend/app/api/deps.py

@@ -6,11 +6,11 @@
 from fastapi.security import OAuth2PasswordBearer
 from jwt.exceptions import InvalidTokenError
 from pydantic import ValidationError
-from sqlmodel import Session
+from sqlalchemy.orm import Session
 
 from app.core import security
 from app.core.config import settings
-from app.core.db import engine
+from app.core.db import SessionLocal
 from app.models import TokenPayload, User
 
 reusable_oauth2 = OAuth2PasswordBearer(
@@ -19,8 +19,11 @@
 
 
 def get_db() -> Generator[Session, None, None]:
-    with Session(engine) as session:
-        yield session
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
 
 
 SessionDep = Annotated[Session, Depends(get_db)]

backend/app/api/routes/items.py

@@ -2,7 +2,7 @@
 from typing import Any
 
 from fastapi import APIRouter, HTTPException
-from sqlmodel import func, select
+from sqlalchemy import func, select
 
 from app.api.deps import CurrentUser, SessionDep
 from app.models import Item, ItemCreate, ItemPublic, ItemsPublic, ItemUpdate, Message
@@ -20,23 +20,23 @@ def read_items(
 
     if current_user.is_superuser:
         count_statement = select(func.count()).select_from(Item)
-        count = session.exec(count_statement).one()
+        count = session.execute(count_statement).scalar_one()
         statement = select(Item).offset(skip).limit(limit)
-        items = session.exec(statement).all()
+        items = session.execute(statement).scalars().all()
     else:
         count_statement = (
             select(func.count())
             .select_from(Item)
             .where(Item.owner_id == current_user.id)
         )
-        count = session.exec(count_statement).one()
+        count = session.execute(count_statement).scalar_one()
         statement = (
             select(Item)
             .where(Item.owner_id == current_user.id)
             .offset(skip)
             .limit(limit)
         )
-        items = session.exec(statement).all()
+        items = session.execute(statement).scalars().all()
 
     return ItemsPublic(data=items, count=count)
 
@@ -61,7 +61,8 @@ def create_item(
     """
     Create new item.
     """
-    item = Item.model_validate(item_in, update={"owner_id": current_user.id})
+    item_data = item_in.model_dump()
+    item = Item(**item_data, owner_id=current_user.id)
     session.add(item)
     session.commit()
     session.refresh(item)
@@ -85,7 +86,8 @@ def update_item(
     if not current_user.is_superuser and (item.owner_id != current_user.id):
         raise HTTPException(status_code=400, detail="Not enough permissions")
     update_dict = item_in.model_dump(exclude_unset=True)
-    item.sqlmodel_update(update_dict)
+    for field, value in update_dict.items():
+        setattr(item, field, value)
     session.add(item)
     session.commit()
     session.refresh(item)

backend/app/api/routes/users.py

@@ -2,7 +2,7 @@
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
-from sqlmodel import col, delete, func, select
+from sqlalchemy import func, select, delete
 
 from app import crud
 from app.api.deps import (
@@ -40,10 +40,10 @@ def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
     """
 
     count_statement = select(func.count()).select_from(User)
-    count = session.exec(count_statement).one()
+    count = session.execute(count_statement).scalar_one()
 
     statement = select(User).offset(skip).limit(limit)
-    users = session.exec(statement).all()
+    users = session.execute(statement).scalars().all()
 
     return UsersPublic(data=users, count=count)
 
@@ -219,8 +219,8 @@ def delete_user(
         raise HTTPException(
             status_code=403, detail="Super users are not allowed to delete themselves"
         )
-    statement = delete(Item).where(col(Item.owner_id) == user_id)
-    session.exec(statement)  # type: ignore
+    statement = delete(Item).where(Item.owner_id == user_id)
+    session.execute(statement)
     session.delete(user)
     session.commit()
     return Message(message="User deleted successfully")

backend/app/backend_pre_start.py

@@ -1,7 +1,7 @@
 import logging
 
-from sqlalchemy import Engine
-from sqlmodel import Session, select
+from sqlalchemy import Engine, select
+from sqlalchemy.orm import Session
 from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed
 
 from app.core.db import engine
@@ -23,7 +23,7 @@ def init(db_engine: Engine) -> None:
     try:
         with Session(db_engine) as session:
             # Try to create session to check if DB is awake
-            session.exec(select(1))
+            session.execute(select(1))
     except Exception as e:
         logger.error(e)
         raise e

backend/app/core/db.py

@@ -1,29 +1,31 @@
-from sqlmodel import Session, create_engine, select
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import sessionmaker
 
 from app import crud
 from app.core.config import settings
-from app.models import User, UserCreate
+from app.models import Base, User, UserCreate
 
 engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 
-# make sure all SQLModel models are imported (app.models) before initializing DB
-# otherwise, SQLModel might fail to initialize relationships properly
+# make sure all SQLAlchemy models are imported (app.models) before initializing DB
+# otherwise, SQLAlchemy might fail to initialize relationships properly
 # for more details: https://github.com/fastapi/full-stack-fastapi-template/issues/28
 
 
-def init_db(session: Session) -> None:
+def init_db(session: SessionLocal) -> None:
     # Tables should be created with Alembic migrations
     # But if you don't want to use migrations, create
     # the tables un-commenting the next lines
-    # from sqlmodel import SQLModel
+    # Base.metadata.create_all(bind=engine)
 
     # This works because the models are already imported and registered from app.models
-    # SQLModel.metadata.create_all(engine)
-
-    user = session.exec(
+
+    result = session.execute(
         select(User).where(User.email == settings.FIRST_SUPERUSER)
-    ).first()
+    )
+    user = result.scalar_one_or_none()
     if not user:
         user_in = UserCreate(
             email=settings.FIRST_SUPERUSER,

backend/app/crud.py

@@ -1,15 +1,21 @@
 import uuid
 from typing import Any
 
-from sqlmodel import Session, select
+from sqlalchemy.orm import Session
+from sqlalchemy import select
 
 from app.core.security import get_password_hash, verify_password
 from app.models import Item, ItemCreate, User, UserCreate, UserUpdate
 
 
 def create_user(*, session: Session, user_create: UserCreate) -> User:
-    db_obj = User.model_validate(
-        user_create, update={"hashed_password": get_password_hash(user_create.password)}
+    user_data = user_create.model_dump()
+    user_data.pop("password", None)  # Remove password from user data
+    hashed_password = get_password_hash(user_create.password)
+
+    db_obj = User(
+        **user_data,
+        hashed_password=hashed_password
     )
     session.add(db_obj)
     session.commit()
@@ -19,12 +25,15 @@ def create_user(*, session: Session, user_create: UserCreate) -> User:
 
 def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any:
     user_data = user_in.model_dump(exclude_unset=True)
-    extra_data = {}
+
     if "password" in user_data:
-        password = user_data["password"]
+        password = user_data.pop("password")
         hashed_password = get_password_hash(password)
-        extra_data["hashed_password"] = hashed_password
-    db_user.sqlmodel_update(user_data, update=extra_data)
+        user_data["hashed_password"] = hashed_password
+
+    for field, value in user_data.items():
+        setattr(db_user, field, value)
+
     session.add(db_user)
     session.commit()
     session.refresh(db_user)
@@ -33,8 +42,8 @@ def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any:
 
 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
+    result = session.execute(statement)
+    return result.scalar_one_or_none()
 
 
 def authenticate(*, session: Session, email: str, password: str) -> User | None:
@@ -47,7 +56,11 @@ def authenticate(*, session: Session, email: str, password: str) -> User | None:
 
 
 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})
+    item_data = item_in.model_dump()
+    db_item = Item(
+        **item_data,
+        owner_id=owner_id
+    )
     session.add(db_item)
     session.commit()
     session.refresh(db_item)

backend/app/initial_data.py

@@ -1,16 +1,17 @@
 import logging
 
-from sqlmodel import Session
-
-from app.core.db import engine, init_db
+from app.core.db import SessionLocal, init_db
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 
 def init() -> None:
-    with Session(engine) as session:
-        init_db(session)
+    db = SessionLocal()
+    try:
+        init_db(db)
+    finally:
+        db.close()
 
 
 def main() -> None: