mzaxd
6eee3f885f
Deploy to Staging / deploy (push) Waiting to run
Lint Backend / lint-backend (push) Waiting to run
Playwright Tests / changes (push) Waiting to run
Playwright Tests / test-playwright (1, 4) (push) Blocked by required conditions
Playwright Tests / test-playwright (2, 4) (push) Blocked by required conditions
Playwright Tests / test-playwright (3, 4) (push) Blocked by required conditions
Playwright Tests / test-playwright (4, 4) (push) Blocked by required conditions
Playwright Tests / merge-playwright-reports (push) Blocked by required conditions
Playwright Tests / alls-green-playwright (push) Blocked by required conditions
Test Backend / test-backend (push) Waiting to run
Test Docker Compose / test-docker-compose (push) Waiting to run
36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
from typing import Any
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from app.core.security import get_password_hash
|
|
from app.models import 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)}
|
|
)
|
|
session.add(db_obj)
|
|
session.commit()
|
|
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 = {}
|
|
if "password" in user_data:
|
|
password = user_data["password"]
|
|
hashed_password = get_password_hash(password)
|
|
extra_data["hashed_password"] = hashed_password
|
|
db_user.sqlmodel_update(user_data, update=extra_data)
|
|
session.add(db_user)
|
|
session.commit()
|
|
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 |