RMM/crud.py

from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError, IntegrityError
import models, schemas
from utils import hash_password, encrypt_data, decrypt_data


def create_user(db: Session, user: schemas.UserCreate):
    """Создает нового пользователя в базе данных на основе предоставленных данных."""
    try:
        hashed_password = hash_password(user.password)
        encrypted_email = encrypt_data(user.email)
        db_user = models.User(name=user.name, email=encrypted_email, password=hashed_password)
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        db_user.email = decrypt_data(db_user.email)
        return db_user
    except IntegrityError:
        db.rollback()
        raise ValueError("User with this email already exists")
    except SQLAlchemyError:
        db.rollback()
        raise ConnectionError("Database error while creating user")


def get_users(db: Session):
    """Получает список всех пользователей из базы данных."""
    try:
        users = db.query(models.User).all()
        for user in users:
            user.email = decrypt_data(user.email)
        return users
    except SQLAlchemyError:
        raise ConnectionError("Database error while fetching users")


def get_user(db: Session, user_id: int):
    """Получает информацию о пользователе по его ID из базы данных."""
    try:
        user = db.query(models.User).filter(models.User.id == user_id).first()
        if not user:
            raise LookupError("User not found")
        user.email = decrypt_data(user.email)
        return user
    except SQLAlchemyError:
        raise ConnectionError("Database error while fetching user")


def delete_user(db: Session, user_id: int):
    """Удаляет пользователя по его ID из базы данных."""
    try:
        user = db.query(models.User).filter(models.User.id == user_id).first()
        if not user:
            raise LookupError("User not found")

        db.delete(user)
        db.commit()
        return user
    except SQLAlchemyError:
        db.rollback()
        raise ConnectionError("Database error while deleting user")


def update_user(db: Session, user_id: int, user_update: schemas.UserCreate):
    """Обновляет данные пользователя по его ID."""
    try:
        user = db.query(models.User).filter(models.User.id == user_id).first()
        if not user:
            raise LookupError("User not found")

        user.name = user_update.name
        user.email = encrypt_data(user_update.email)
        user.password = hash_password(user_update.password)

        db.commit()
        db.refresh(user)
        user.email = decrypt_data(user.email)
        return user
    except SQLAlchemyError:
        db.rollback()
        raise ConnectionError("Database error while updating user")


def create_post(db: Session, post: schemas.PostCreate):
    """Создает новый пост в базе данных, проверяя существование пользователя."""
    try:
        user = db.query(models.User).filter(models.User.id == post.user_id).first()
        if not user:
            raise LookupError("User does not exist")

        db_post = models.Post(**post.dict())
        db.add(db_post)
        db.commit()
        db.refresh(db_post)
        return db_post

    except IntegrityError:
        db.rollback()
        raise ValueError("Invalid post data")
    except SQLAlchemyError:
        db.rollback()
        raise ConnectionError("Database error while creating post")


def get_posts(db: Session):
    """Получает список всех постов из базы данных."""
    try:
        return db.query(models.Post).all()
    except SQLAlchemyError:
        raise ConnectionError("Database error while fetching posts")


def get_post(db: Session, post_id: int):
    """Получает информацию о посте по его ID из базы данных."""
    try:
        post = db.query(models.Post).filter(models.Post.id == post_id).first()
        if not post:
            raise LookupError("Post not found")
        return post
    except SQLAlchemyError:
        raise ConnectionError("Database error while fetching post")


def update_post(db: Session, post_id: int, post_update: schemas.PostCreate):
    """Обновляет данные поста по его ID."""
    try:
        post = db.query(models.Post).filter(models.Post.id == post_id).first()
        if not post:
            raise LookupError("Post not found")

        for key, value in post_update.dict().items():
            setattr(post, key, value)

        db.commit()
        db.refresh(post)
        return post
    except SQLAlchemyError:
        db.rollback()
        raise ConnectionError("Database error while updating post")