Major addition. User register, login and update is now functional on frontend, removed UUID5 for UUID4 for users and many more changes
This commit is contained in:
1 parent
d87c280109
commit
d0690f2f06
67 files changed
+3389
-1066
No files matched your search
@@ -1,5 +1,6 @@
|
||||
import logging
|
||||
from typing import Optional
|
||||
from uuid import UUID, uuid4
|
||||
from uuid import uuid4
|
||||
|
||||
from sqlmodel import Session, select
|
||||
|
||||
@@ -7,12 +8,19 @@ from app.database.models.shop_model import Shop, ShopStatus
|
||||
from app.database.models.user_model import User
|
||||
from app.schemas.shop_schemas import ShopCreate
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def get_shop_id_from_uuid(session: Session, shop_id: int) -> Optional[UUID]:
|
||||
def get_shop_by_id(session: Session, shop_id: int) -> Optional[Shop]:
|
||||
logger.debug("Getting shop by shop_id - %d", id)
|
||||
stmt = select(Shop).where(Shop.id == shop_id)
|
||||
db_shop = session.exec(stmt).one_or_none()
|
||||
return db_shop
|
||||
|
||||
def get_shop_by_uuid(session: Session, shop_uuid: str) -> Optional[Shop]:
|
||||
logger.debug("Getting shop by UUID - %s" , shop_uuid)
|
||||
stmt = select(Shop).where(Shop.uuid == shop_uuid)
|
||||
db_shop = session.exec(stmt).one_or_none()
|
||||
return db_shop
|
||||
|
||||
def create_shop(session: Session, shop_data: ShopCreate, creator: User) -> None:
|
||||
shop_uuid = uuid4()
|
||||
|
||||
@@ -1,58 +1,83 @@
|
||||
import logging
|
||||
from typing import Optional
|
||||
from uuid import UUID
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from sqlmodel import Session, select
|
||||
from fastapi import HTTPException, status
|
||||
from sqlmodel import Session, select, and_
|
||||
|
||||
from app.core.security import get_password_hash, verify_password
|
||||
from app.crud.shop_crud import get_shop_id_from_uuid
|
||||
from app.crud.shop_crud import get_shop_by_uuid
|
||||
from app.database.models.user_model import User, UserRole
|
||||
from app.schemas.user_schemas import UserRegister
|
||||
from app.schemas.user_schemas import UserRegister, UserUpdate
|
||||
from app.utils.models import generate_user_uuid5
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_user_by_generated_uuid(session: Session, email: str, shop_uuid: Optional[UUID]) -> Optional[User]:
|
||||
logger.debug("Getting shop id by UUID - %s", shop_uuid)
|
||||
shop_id = get_shop_id_from_uuid(session, shop_uuid)
|
||||
logger.debug("Generating user UUID5")
|
||||
user_uuid = generate_user_uuid5(email, shop_id)
|
||||
stmt = select(User).where(User.uuid == user_uuid)
|
||||
logger.debug("Executing select query")
|
||||
db_user = session.exec(stmt).one_or_none()
|
||||
return db_user
|
||||
|
||||
def create_user(session: Session, user_register: UserRegister, shop_uuid: Optional[UUID], user_role: UserRole):
|
||||
logger.debug("Getting shop id by UUID - %s", shop_uuid)
|
||||
shop_id = get_shop_id_from_uuid(session, shop_uuid)
|
||||
logger.debug("Generating user UUID5")
|
||||
user_uuid = generate_user_uuid5(user_register.email, shop_id)
|
||||
if shop_uuid:
|
||||
logger.debug("Fetching shop by UUID")
|
||||
shop_id = get_shop_by_uuid(session, shop_uuid).id
|
||||
else:
|
||||
logger.debug("No shop UUID provided -> Owner account is being created")
|
||||
shop_id = None
|
||||
|
||||
logger.debug("Hashing password")
|
||||
hashed_password = get_password_hash(user_register.password)
|
||||
|
||||
new_user = User(
|
||||
uuid=user_uuid,
|
||||
uuid=uuid4(),
|
||||
shop_id=shop_id,
|
||||
email=user_register.email,
|
||||
username=user_register.username,
|
||||
phone_number=user_register.phone_number,
|
||||
user_role=user_role,
|
||||
password=hashed_password
|
||||
)
|
||||
)
|
||||
|
||||
logger.debug("Inserting new user")
|
||||
session.add(new_user)
|
||||
session.commit()
|
||||
|
||||
|
||||
def update_user(session: Session, user_update: UserUpdate, current_user: User):
|
||||
|
||||
current_user.email = user_update.email
|
||||
current_user.username = user_update.username
|
||||
current_user.phone_number = user_update.phone_number
|
||||
current_user.first_name = user_update.first_name
|
||||
current_user.last_name = user_update.last_name
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def get_user_by_uuid(session: Session, email: str, shop_uuid: Optional[UUID]) -> Optional[User]:
|
||||
if shop_uuid:
|
||||
shop_id = get_shop_by_uuid(session, shop_uuid).id
|
||||
else:
|
||||
shop_id = None
|
||||
|
||||
stmt = select(User).where(and_(
|
||||
User.email == email,
|
||||
User.shop_id == shop_id
|
||||
))
|
||||
logger.debug("Executing select query")
|
||||
db_user = session.exec(stmt).one_or_none()
|
||||
return db_user
|
||||
|
||||
|
||||
def authenticate(session: Session, email: str, password: str, shop_uuid: Optional[int]) -> Optional[User]:
|
||||
logger.debug("Getting shop id by UUID - %s", shop_uuid)
|
||||
shop_id = get_shop_id_from_uuid(session, shop_uuid)
|
||||
if shop_uuid:
|
||||
shop_id = get_shop_by_uuid(session, shop_uuid).id
|
||||
else:
|
||||
shop_id = None
|
||||
|
||||
logger.debug("Fetching user from db by email - %s", email)
|
||||
db_user = get_user_by_generated_uuid(session, email, shop_id)
|
||||
db_user = get_user_by_uuid(session, email, shop_id)
|
||||
if db_user is None:
|
||||
logger.warn("Didn't find User with email=%s for shop=%s", email, shop_uuid)
|
||||
logger.warning("Didn't find User with email=%s for shop=%s", email, shop_uuid)
|
||||
return None
|
||||
if not verify_password(plain_password=password, hashed_password=db_user.password):
|
||||
logger.warn("Found user with email=%s for shop=%s", email, shop_uuid)
|
||||
logger.warning("Found user with email=%s for shop=%s", email, shop_uuid)
|
||||
return None
|
||||
return db_user
|
||||
Reference in new issue
Block a user