47 lines
1.9 KiB
Python
47 lines
1.9 KiB
Python
from typing import Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select
|
|
import bcrypt
|
|
from app_core.db.database import engine, Base, SessionLocal
|
|
from app_core.db.models import User
|
|
|
|
# Create tables on import
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
class AuthService:
|
|
def __init__(self) -> None:
|
|
self._session_factory = SessionLocal
|
|
|
|
def _hash_password(self, raw_password: str) -> str:
|
|
salt = bcrypt.gensalt()
|
|
return bcrypt.hashpw(raw_password.encode("utf-8"), salt).decode("utf-8")
|
|
|
|
def _verify_password(self, raw_password: str, hashed: str) -> bool:
|
|
try:
|
|
return bcrypt.checkpw(raw_password.encode("utf-8"), hashed.encode("utf-8"))
|
|
except Exception:
|
|
return False
|
|
|
|
def signup(self, email: str, password: str) -> tuple[bool, str]:
|
|
email = email.strip().lower()
|
|
if not email or not password:
|
|
return False, "Email and password are required."
|
|
with self._session_factory() as db: # type: Session
|
|
exists = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
|
|
if exists:
|
|
return False, "Email already registered."
|
|
user = User(email=email, password_hash=self._hash_password(password))
|
|
db.add(user)
|
|
db.commit()
|
|
return True, "Account created. Please login."
|
|
|
|
def login(self, email: str, password: str) -> tuple[bool, Optional[dict], str]:
|
|
email = email.strip().lower()
|
|
if not email or not password:
|
|
return False, None, "Email and password are required."
|
|
with self._session_factory() as db: # type: Session
|
|
user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
|
|
if not user or not self._verify_password(password, user.password_hash):
|
|
return False, None, "Invalid credentials."
|
|
return True, {"id": user.id, "email": user.email}, "Login successful."
|