57 lines
1.4 KiB
Python
57 lines
1.4 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .auth import hash_password, verify_password
|
|
from .models import Address, User
|
|
from .schemas import AddressCreate, UserCreate
|
|
|
|
|
|
def get_user_by_id(db: Session, user_id: int) -> User | None:
|
|
return db.get(User, user_id)
|
|
|
|
|
|
def get_user_by_email(db: Session, email: str) -> User | None:
|
|
statement = select(User).where(User.email == email)
|
|
return db.scalar(statement)
|
|
|
|
|
|
def create_user(db: Session, user_in: UserCreate) -> User:
|
|
user = User(
|
|
name=user_in.name,
|
|
email=user_in.email,
|
|
password_hash=hash_password(user_in.password),
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
def authenticate_user(db: Session, email: str, password: str) -> User | None:
|
|
user = get_user_by_email(db, email)
|
|
if user is None:
|
|
return None
|
|
if not verify_password(password, user.password_hash):
|
|
return None
|
|
|
|
user.last_login_at = datetime.now(timezone.utc)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
def create_address(db: Session, user: User, address_in: AddressCreate) -> Address:
|
|
address = Address(
|
|
user_id=user.id,
|
|
house_number=address_in.house_number,
|
|
colonia=address_in.colonia,
|
|
street=address_in.street,
|
|
)
|
|
db.add(address)
|
|
db.commit()
|
|
db.refresh(address)
|
|
return address
|