246b4e058e
Added complete backend infrastructure with: - Authentication system with OAuth (Google, GitHub, Apple) - Stripe payment processing with subscription management - Testimonials management API - Usage statistics tracking - Email communication services - Health monitoring endpoints - Database migrations with Alembic - Comprehensive API documentation All APIs are production-ready with proper error handling, security measures, and environment variable configuration. Co-Authored-By: Claude <noreply@anthropic.com>
71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
from typing import Optional
|
|
from sqlalchemy.orm import Session
|
|
from app.models.user import User
|
|
from app.schemas.user import UserCreate, UserUpdate
|
|
from app.core.security import get_password_hash, verify_password
|
|
|
|
class UserService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def get_user(self, user_id: int) -> Optional[User]:
|
|
return self.db.query(User).filter(User.id == user_id).first()
|
|
|
|
def get_user_by_email(self, email: str) -> Optional[User]:
|
|
return self.db.query(User).filter(User.email == email).first()
|
|
|
|
def get_user_by_username(self, username: str) -> Optional[User]:
|
|
return self.db.query(User).filter(User.username == username).first()
|
|
|
|
def create_user(self, user: UserCreate) -> User:
|
|
hashed_password = get_password_hash(user.password)
|
|
db_user = User(
|
|
email=user.email,
|
|
username=user.username,
|
|
full_name=user.full_name,
|
|
hashed_password=hashed_password
|
|
)
|
|
self.db.add(db_user)
|
|
self.db.commit()
|
|
self.db.refresh(db_user)
|
|
return db_user
|
|
|
|
def create_oauth_user(self, email: str, full_name: str, provider: str, provider_id: str, avatar_url: str = None) -> User:
|
|
db_user = User(
|
|
email=email,
|
|
full_name=full_name,
|
|
avatar_url=avatar_url,
|
|
is_verified=True
|
|
)
|
|
|
|
if provider == "google":
|
|
db_user.google_id = provider_id
|
|
elif provider == "github":
|
|
db_user.github_id = provider_id
|
|
elif provider == "apple":
|
|
db_user.apple_id = provider_id
|
|
|
|
self.db.add(db_user)
|
|
self.db.commit()
|
|
self.db.refresh(db_user)
|
|
return db_user
|
|
|
|
def update_user(self, user_id: int, user_update: UserUpdate) -> Optional[User]:
|
|
db_user = self.get_user(user_id)
|
|
if not db_user:
|
|
return None
|
|
|
|
for field, value in user_update.dict(exclude_unset=True).items():
|
|
setattr(db_user, field, value)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(db_user)
|
|
return db_user
|
|
|
|
def authenticate_user(self, email: str, password: str) -> Optional[User]:
|
|
user = self.get_user_by_email(email)
|
|
if not user or not user.hashed_password:
|
|
return None
|
|
if not verify_password(password, user.hashed_password):
|
|
return None
|
|
return user |