SQLAlchemy 2.0 (#26)

* SQLAlchemy 2.0

Author: grigoriev-semyon

auth_backend/auth_plugins/email.py

@@ -16,6 +16,7 @@
     send_confirmation_email,
     send_change_password_confirmation,
     send_changes_password_notification,
+    send_reset_email,
 )
 from .auth_method import AuthMethodMeta, Session
 from fastapi.background import BackgroundTasks
@@ -133,7 +134,7 @@ async def _login(user_inp: EmailLogin) -> Session:
         ):
             raise AuthFailed(error="Incorrect login or password")
         db.session.add(user_session := UserSession(user_id=query.user.id, token=random_string()))
-        db.session.flush()
+        db.session.commit()
         return Session(
             user_id=user_session.user_id, token=user_session.token, id=user_session.id, expires=user_session.expires
         )
@@ -155,13 +156,14 @@ async def _add_to_db(user_inp: EmailRegister, confirmation_token: str, user: Use
         )
         db.session.flush()
 
+
     @staticmethod
     async def _change_confirmation_link(user: User, confirmation_token: str) -> None:
         if user.auth_methods.confirmed.value == "true":
             raise AlreadyExists(User, user.id)
         else:
             user.auth_methods.confirmation_token.value = confirmation_token
-            db.session.flush()
+
 
     @staticmethod
     async def _get_user_by_token_and_id(id: int, token: str) -> User:
@@ -198,6 +200,7 @@ async def _register(
                 to_addr=user_inp.email,
                 link=f"{settings.APPLICATION_HOST}/email/approve?token={confirmation_token}",
             )
+            db.session.commit()
             return ResponseModel(status="Success", message="Email confirmation link sent")
         if user_inp.user_id and token:
             user = await Email._get_user_by_token_and_id(user_inp.user_id, token)
@@ -211,6 +214,7 @@ async def _register(
             to_addr=user_inp.email,
             link=f"{settings.APPLICATION_HOST}/email/approve?token={confirmation_token}",
         )
+        db.session.commit()
         raise HTTPException(
             status_code=201, detail=ResponseModel(status="Success", message="Email confirmation link sent").json()
         )
@@ -238,8 +242,8 @@ async def _approve_email(token: str) -> ResponseModel:
         )
         if not auth_method:
             raise HTTPException(status_code=403, detail=ResponseModel(status="Error", message="Incorrect link").json())
-        auth_method.user.auth_methods.confirmed.value = True
-        db.session.flush()
+        auth_method.user.auth_methods.confirmed.value = "true"
+        db.session.commit()
         return ResponseModel(status="Success", message="Email approved")
 
     @staticmethod
@@ -269,12 +273,12 @@ async def _request_reset_email(
             user_id=session.user_id, auth_method=Email.get_name(), param="tmp_email_confirmation_token", value=token
         )
         db.session.add_all([tmp_email, tmp_email_confirmation_token])
-        db.session.flush()
         background_tasks.add_task(
-            send_confirmation_email,
+            send_reset_email,
             to_addr=scheme.email,
             link=f"{settings.APPLICATION_HOST}/email/reset/email/{session.user_id}?token={token}&email={scheme.email}",
         )
+        db.session.commit()
         return ResponseModel(status="Success", message="Email confirmation link sent")
 
     @staticmethod
@@ -295,7 +299,7 @@ async def _reset_email(user_id: int, token: str, email: str):
         user.auth_methods.email.value = user.auth_methods.tmp_email.value
         db.session.delete(user.auth_methods.tmp_email_confirmation_token)
         db.session.delete(user.auth_methods.tmp_email)
-        db.session.flush()
+        db.session.commit()
         return ResponseModel(status="Success", message="Email successfully changed")
 
     @staticmethod
@@ -326,8 +330,8 @@ async def _request_reset_password(
                 )
             session.user.auth_methods.hashed_password.value = Email._hash_password(schema.new_password, salt)
             session.user.auth_methods.salt.value = salt
-            db.session.flush()
             background_tasks.add_task(send_changes_password_notification, session.user.auth_methods.email.value)
+            db.session.commit()
             return ResponseModel(status="Success", message="Password has been successfully changed")
         elif not token and not schema.password and not schema.new_password:
             user: User = db.session.query(User).get(user_id)
@@ -345,7 +349,7 @@ async def _request_reset_password(
             db.session.add(
                 AuthMethod(user_id=user_id, auth_method=Email.get_name(), param="reset_token", value=random_string())
             )
-            db.session.flush()
+            db.session.commit()
             background_tasks.add_task(
                 send_change_password_confirmation,
                 user.auth_methods.email.value,
@@ -372,5 +376,5 @@ async def _reset_password(
         user.auth_methods.hashed_password.value = Email._hash_password(schema.new_password, salt)
         user.auth_methods.salt.value = salt
         db.session.delete(user.auth_methods.reset_token)
-        db.session.flush()
+        db.session.commit()
         return ResponseModel(status="Success", message="Password has been successfully changed")

auth_backend/models/db.py

@@ -3,6 +3,8 @@
 import datetime
 
 import sqlalchemy.orm
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy import String, Integer, ForeignKey, DateTime
 from sqlalchemy.ext.hybrid import hybrid_property
 
 from auth_backend.models.base import Base
@@ -32,10 +34,10 @@ def __new__(cls, methods: list[AuthMethod], *args, **kwargs):
 
 class User(Base):
 
-    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
 
-    _auth_methods: list[AuthMethod] = sqlalchemy.orm.relationship("AuthMethod", foreign_keys="AuthMethod.user_id")
-    sessions: list[UserSession] = sqlalchemy.orm.relationship("UserSession", foreign_keys="UserSession.user_id")
+    _auth_methods: Mapped[list["AuthMethod"]] = relationship("AuthMethod", foreign_keys="AuthMethod.user_id")
+    sessions: Mapped[list["UserSession"]] = relationship("UserSession", foreign_keys="UserSession.user_id")
 
     @hybrid_property
     def auth_methods(self) -> ParamDict:
@@ -48,22 +50,22 @@ def auth_methods(self) -> ParamDict:
 
 
 class AuthMethod(Base):
-    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
-    user_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey("user.id"))
-    auth_method = sqlalchemy.Column(sqlalchemy.String)
-    param = sqlalchemy.Column(sqlalchemy.String)
-    value = sqlalchemy.Column(sqlalchemy.String)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"))
+    auth_method: Mapped[str] = mapped_column(String)
+    param: Mapped[str] = mapped_column(String)
+    value: Mapped[str] = mapped_column(String)
 
-    user: User = sqlalchemy.orm.relationship("User", foreign_keys=[user_id], back_populates="_auth_methods")
+    user: Mapped["User"] = relationship("User", foreign_keys=[user_id], back_populates="_auth_methods")
 
 
 class UserSession(Base):
-    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
-    user_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey("user.id"))
-    expires = sqlalchemy.Column(sqlalchemy.DateTime, default=datetime.datetime.utcnow() + datetime.timedelta(days=7))
-    token = sqlalchemy.Column(sqlalchemy.String, unique=True)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    user_id: Mapped[int] = mapped_column(Integer, sqlalchemy.ForeignKey("user.id"))
+    expires: Mapped[datetime.datetime] = mapped_column(DateTime, default=datetime.datetime.utcnow() + datetime.timedelta(days=7))
+    token: Mapped[str] = mapped_column(String, unique=True)
 
-    user: User = sqlalchemy.orm.relationship("User", foreign_keys=[user_id], back_populates="sessions")
+    user: Mapped["User"] = relationship("User", foreign_keys=[user_id], back_populates="sessions")
 
     @hybrid_property
     def expired(self):

auth_backend/routes/base.py

@@ -12,7 +12,7 @@
 
 
 app.add_middleware(
-    DBSessionMiddleware, db_url=settings.DB_DSN, session_args={"autocommit": True}, engine_args={"pool_pre_ping": True}
+    DBSessionMiddleware, db_url=settings.DB_DSN,  engine_args={"pool_pre_ping": True}
 )
 
 app.add_middleware(

auth_backend/routes/user_session.py

@@ -21,7 +21,7 @@ async def logout(token: str = Header(min_length=1)) -> JSONResponse:
     if session.expired:
         raise SessionExpired(session.token)
     session.expires = datetime.utcnow()
-    db.session.flush()
+    db.session.commit()
     return JSONResponse(status_code=200, content=ResponseModel(status="Success", message="Logout successful").json())
 
 

tests/conftest.py

@@ -18,6 +18,7 @@ def client():
     auth_backend.auth_plugins.email.send_confirmation_email = Mock(return_value=None)
     auth_backend.auth_plugins.email.send_change_password_confirmation = Mock(return_value=None)
     auth_backend.auth_plugins.email.send_changes_password_notification = Mock(return_value=None)
+    auth_backend.auth_plugins.email.send_reset_email = Mock(return_value=None)
     client = TestClient(app)
     yield client
 
@@ -26,7 +27,7 @@ def client():
 def dbsession():
     settings = get_settings()
     engine = create_engine(settings.DB_DSN)
-    TestingSessionLocal = sessionmaker(autocommit=True, autoflush=False, bind=engine)
+    TestingSessionLocal = sessionmaker(bind=engine)
     return TestingSessionLocal()
 
 
@@ -42,7 +43,7 @@ def user_id(client: TestClient, dbsession):
     for row in dbsession.query(AuthMethod).filter(AuthMethod.user_id == db_user.user_id).all():
         dbsession.delete(row)
     dbsession.delete(dbsession.query(User).filter(User.id == db_user.user_id).one())
-    dbsession.flush()
+    dbsession.commit()
 
 
 @pytest.fixture()
@@ -73,4 +74,4 @@ def user(client: TestClient, dbsession):
     for row in dbsession.query(AuthMethod).filter(AuthMethod.user_id == db_user.user_id).all():
         dbsession.delete(row)
     dbsession.delete(dbsession.query(User).filter(User.id == db_user.user_id).one())
-    dbsession.flush()
+    dbsession.commit()

tests/test_routes/test_change_email.py

@@ -1,3 +1,5 @@
+import datetime
+
 from fastapi.testclient import TestClient
 from sqlalchemy.orm import Session
 from starlette import status
@@ -15,7 +17,8 @@ def test_main_scenario(client: TestClient, dbsession: Session, user):
         .one()
         .value
     )
-    response = client.post(f"{url}request", json={"email": "changed@mail.com"}, headers={"token": login["token"]})
+    tmp_email = f"changed{datetime.datetime.utcnow()}@mail.com"
+    response = client.post(f"{url}request", json={"email": tmp_email}, headers={"token": login["token"]})
     assert response.status_code == status.HTTP_200_OK
 
     conf_token_2 = (
@@ -38,7 +41,7 @@ def test_main_scenario(client: TestClient, dbsession: Session, user):
     response = client.post(f"/email/login", json=body)
     assert response.status_code == status.HTTP_200_OK
 
-    response = client.post(f"/email/login", json={"email": "changed@mail.com", "password": body["password"]})
+    response = client.post(f"/email/login", json={"email": tmp_email, "password": body["password"]})
     assert response.status_code == status.HTTP_401_UNAUTHORIZED
 
     response = client.get(f"{url}{user_id}?token={conf_token_1}&email=changed@mail.com")
@@ -47,13 +50,13 @@ def test_main_scenario(client: TestClient, dbsession: Session, user):
     response = client.get(f"{url}{user_id}?token={tmp_token}&email=wrong@mail.com")
     assert response.status_code == status.HTTP_403_FORBIDDEN
 
-    response = client.get(f"{url}{user_id}?token={tmp_token}&email=changed@mail.com")
+    response = client.get(f"{url}{user_id}?token={tmp_token}&email={tmp_email}")
     assert response.status_code == status.HTTP_200_OK
 
     response = client.post(f"/email/login", json=body)
     assert response.status_code == status.HTTP_401_UNAUTHORIZED
 
-    response = client.post(f"/email/login", json={"email": "changed@mail.com", "password": body["password"]})
+    response = client.post(f"/email/login", json={"email": tmp_email, "password": body["password"]})
     assert response.status_code == status.HTTP_200_OK
 
 

tests/test_routes/test_login.py

@@ -67,7 +67,7 @@ def test_incorrect_data(client: TestClient, dbsession: Session):
     for row in dbsession.query(AuthMethod).filter(AuthMethod.user_id == id).all():
         dbsession.delete(row)
     dbsession.delete(dbsession.query(User).filter(User.id == id).one())
-    dbsession.flush()
+    dbsession.commit()
 
 
 def test_check_token(client: TestClient, user, dbsession: Session):

tests/test_routes/test_logout.py

@@ -41,7 +41,7 @@ def test_main_scenario(client: TestClient, dbsession: Session):
         dbsession.delete(row)
     dbsession.delete(dbsession.query(UserSession).filter(UserSession.user_id == id).one())
     dbsession.delete(dbsession.query(User).filter(User.id == id).one())
-    dbsession.flush()
+    dbsession.commit()
 
 
 def test_without_token(client: TestClient, dbsession: Session):

tests/test_routes/test_registration.py

@@ -63,7 +63,7 @@ def test_main_scenario(client: TestClient, dbsession: Session):
     for row in dbsession.query(AuthMethod).filter(AuthMethod.user_id == db_user.user_id).all():
         dbsession.delete(row)
     dbsession.delete(dbsession.query(User).filter(User.id == db_user.user_id).one())
-    dbsession.flush()
+    dbsession.commit()
 
 
 def test_repeated_registration_case(client: TestClient, dbsession: Session):
@@ -105,4 +105,4 @@ def test_repeated_registration_case(client: TestClient, dbsession: Session):
     for row in dbsession.query(AuthMethod).filter(AuthMethod.user_id == user_id).all():
         dbsession.delete(row)
     dbsession.delete(dbsession.query(User).filter(User.id == user_id).one())
-    dbsession.flush()
+    dbsession.commit()