# app/crud/list.py from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select from sqlalchemy.orm import selectinload, joinedload from sqlalchemy import or_, and_, delete as sql_delete, func as sql_func, desc from sqlalchemy.exc import SQLAlchemyError, IntegrityError, OperationalError from typing import Optional, List as PyList from app.schemas.list import ListStatus from app.models import List as ListModel, UserGroup as UserGroupModel, Item as ItemModel from app.schemas.list import ListCreate, ListUpdate from app.core.exceptions import ( ListNotFoundError, ListPermissionError, ListCreatorRequiredError, DatabaseConnectionError, DatabaseIntegrityError, DatabaseQueryError, DatabaseTransactionError ) async def create_list(db: AsyncSession, list_in: ListCreate, creator_id: int) -> ListModel: """Creates a new list record.""" try: async with db.begin(): db_list = ListModel( name=list_in.name, description=list_in.description, group_id=list_in.group_id, created_by_id=creator_id, is_complete=False ) db.add(db_list) await db.flush() await db.refresh(db_list) return db_list except IntegrityError as e: raise DatabaseIntegrityError(f"Failed to create list: {str(e)}") except OperationalError as e: raise DatabaseConnectionError(f"Database connection error: {str(e)}") except SQLAlchemyError as e: raise DatabaseTransactionError(f"Failed to create list: {str(e)}") async def get_lists_for_user(db: AsyncSession, user_id: int) -> PyList[ListModel]: """Gets all lists accessible by a user.""" try: async with db.begin(): group_ids_result = await db.execute( select(UserGroupModel.group_id).where(UserGroupModel.user_id == user_id) ) user_group_ids = group_ids_result.scalars().all() query = select(ListModel).where( or_( and_(ListModel.created_by_id == user_id, ListModel.group_id == None), ListModel.group_id.in_(user_group_ids) ) ).order_by(ListModel.updated_at.desc()) result = await db.execute(query) return result.scalars().all() except OperationalError as e: raise DatabaseConnectionError(f"Failed to connect to database: {str(e)}") except SQLAlchemyError as e: raise DatabaseQueryError(f"Failed to query user lists: {str(e)}") async def get_list_by_id(db: AsyncSession, list_id: int, load_items: bool = False) -> Optional[ListModel]: """Gets a single list by ID, optionally loading its items.""" try: async with db.begin(): query = select(ListModel).where(ListModel.id == list_id) if load_items: query = query.options( selectinload(ListModel.items) .options( joinedload(ItemModel.added_by_user), joinedload(ItemModel.completed_by_user) ) ) result = await db.execute(query) return result.scalars().first() except OperationalError as e: raise DatabaseConnectionError(f"Failed to connect to database: {str(e)}") except SQLAlchemyError as e: raise DatabaseQueryError(f"Failed to query list: {str(e)}") async def update_list(db: AsyncSession, list_db: ListModel, list_in: ListUpdate) -> ListModel: """Updates an existing list record.""" try: async with db.begin(): update_data = list_in.model_dump(exclude_unset=True) for key, value in update_data.items(): setattr(list_db, key, value) db.add(list_db) await db.flush() await db.refresh(list_db) return list_db except IntegrityError as e: raise DatabaseIntegrityError(f"Failed to update list: {str(e)}") except OperationalError as e: raise DatabaseConnectionError(f"Database connection error: {str(e)}") except SQLAlchemyError as e: raise DatabaseTransactionError(f"Failed to update list: {str(e)}") async def delete_list(db: AsyncSession, list_db: ListModel) -> None: """Deletes a list record.""" try: async with db.begin(): await db.delete(list_db) return None except OperationalError as e: raise DatabaseConnectionError(f"Database connection error: {str(e)}") except SQLAlchemyError as e: raise DatabaseTransactionError(f"Failed to delete list: {str(e)}") async def check_list_permission(db: AsyncSession, list_id: int, user_id: int, require_creator: bool = False) -> ListModel: """Fetches a list and verifies user permission.""" try: async with db.begin(): list_db = await get_list_by_id(db, list_id=list_id, load_items=True) if not list_db: raise ListNotFoundError(list_id) is_creator = list_db.created_by_id == user_id if require_creator: if not is_creator: raise ListCreatorRequiredError(list_id, "access") return list_db if is_creator: return list_db if list_db.group_id: from app.crud.group import is_user_member is_member = await is_user_member(db, group_id=list_db.group_id, user_id=user_id) if not is_member: raise ListPermissionError(list_id) return list_db else: raise ListPermissionError(list_id) except OperationalError as e: raise DatabaseConnectionError(f"Failed to connect to database: {str(e)}") except SQLAlchemyError as e: raise DatabaseQueryError(f"Failed to check list permissions: {str(e)}") async def get_list_status(db: AsyncSession, list_id: int) -> ListStatus: """Gets the update timestamps and item count for a list.""" try: async with db.begin(): list_query = select(ListModel.updated_at).where(ListModel.id == list_id) list_result = await db.execute(list_query) list_updated_at = list_result.scalar_one_or_none() if list_updated_at is None: raise ListNotFoundError(list_id) item_status_query = ( select( sql_func.max(ItemModel.updated_at).label("latest_item_updated_at"), sql_func.count(ItemModel.id).label("item_count") ) .where(ItemModel.list_id == list_id) ) item_result = await db.execute(item_status_query) item_status = item_result.first() return ListStatus( list_updated_at=list_updated_at, latest_item_updated_at=item_status.latest_item_updated_at if item_status else None, item_count=item_status.item_count if item_status else 0 ) except OperationalError as e: raise DatabaseConnectionError(f"Failed to connect to database: {str(e)}") except SQLAlchemyError as e: raise DatabaseQueryError(f"Failed to get list status: {str(e)}")