Untitled
unknown
plain_text
a month ago
3.9 kB
3
Indexable
Never
from datetime import datetime from typing import List from sqlalchemy import update from strawberry import mutation, type from app.dependencies.get_db import get_db from app.graphql.schema import Info from app.graphql.types.content_type import ( CreateContentInput, DeleteContentInput, UpdateContentInput, ) from app.models import User from app.models.content import Content @type class UpdateMutationPayload: contents: [Content] success: bool message: str = "" @type class CreateMutationPayload: contents: [Content] success: bool message: str = "" @type class DeleteMutationPayload: contents: [Content] success: bool message: str = "" @type class ContentMutation: @mutation def update_contents( self, info: Info, contents: List[UpdateContentInput] ) -> UpdateMutationPayload: # Use the context manager to manage the database session try: for content in contents: stmt = ( update(Content) .where(Content.id == content.id) .values( type=content.type, text=content.text, link=content.link, order=content.order, updated_date=datetime.now(), # Automatically set to the current time ) ) info.context.session.execute(stmt) info.context.session.commit() return UpdateMutationPayload( contents=Content, success=True, message="Content updated successfully" ) except Exception as e: info.context.session.rollback() return UpdateMutationPayload(success=False, message=f"Update failed: {str(e)}") @mutation def create_contents( self, info: Info, contents: List[CreateContentInput] ) -> CreateMutationPayload: try: for content in contents: new_content = Content( type=content.type, text=content.text, link=content.link, order=content.order, created_date=datetime.now(), updated_date=datetime.now(), ) info.context.session.add(new_content) info.context.session.commit() return CreateMutationPayload( success=True, message="Content created successfully" ) except Exception as e: info.context.session.rollback() return CreateMutationPayload(success=False, message=f"Creation failed: {str(e)}") @mutation def delete_contents( self, info: Info, contents: List[DeleteContentInput] ) -> DeleteMutationPayload: try: total_deleted = 0 for content in contents: stmt = ( update(Content) .where(Content.id == content.id) .values(is_deleted=True) ) result = info.context.session.execute(stmt) total_deleted += result.rowcount info.context.session.commit() if total_deleted > 0: return DeleteMutationPayload( success=True, message=f"{total_deleted} content(s) deleted successfully", ) else: return DeleteMutationPayload( success=False, message="No content was deleted" ) except Exception as e: info.context.session.rollback() return DeleteMutationPayload( success=False, message=f"Deletion failed: {str(e)}" )
Leave a Comment