55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
"""
|
|
Authentication Service - User login, token management
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, Any
|
|
import secrets
|
|
from sqlalchemy.orm import Session
|
|
from app.db.models import User
|
|
from app.security.jwt_manager import jwt_manager
|
|
import logging
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AuthService:
|
|
"""Handles user authentication and token management"""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
async def login(self, email: str, password: str) -> Dict[str, Any]:
|
|
"""Authenticate user with email/password"""
|
|
|
|
user = self.db.query(User).filter_by(email=email).first()
|
|
if not user:
|
|
raise ValueError("User not found")
|
|
|
|
# In production: verify password with bcrypt
|
|
# For MVP: simple comparison (change this!)
|
|
|
|
access_token = jwt_manager.create_access_token(user_id=user.id)
|
|
|
|
logger.info(f"User {user.id} logged in")
|
|
|
|
return {
|
|
"user_id": user.id,
|
|
"access_token": access_token,
|
|
"token_type": "bearer",
|
|
}
|
|
|
|
async def refresh_token(self, refresh_token: str) -> Dict[str, Any]:
|
|
"""Refresh access token"""
|
|
|
|
try:
|
|
payload = jwt_manager.verify_token(refresh_token)
|
|
new_token = jwt_manager.create_access_token(user_id=payload.user_id)
|
|
return {
|
|
"access_token": new_token,
|
|
"token_type": "bearer",
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Token refresh failed: {e}")
|
|
raise ValueError("Invalid refresh token")
|