Untitled
unknown
plain_text
a month ago
2.9 kB
2
Indexable
Never
from app.graphql.schema import Info from app.graphql.types.content_type import UpdateContentInput, CreateContentInput, DeleteContentInput from app.models import User from app.models.content import Content from strawberry import mutation, type from sqlalchemy import update from datetime import datetime from app.dependencies.get_db import get_db from typing import List from app.graphql.schema import Info @type class UpdateContentMutation: @mutation def update_contents(self, info: Info, contents: List[UpdateContentInput]) -> bool: # 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 updated_by=content.updated_by ) ) info.context.session.execute(stmt) info.context.session.commit() return True except Exception as e: info.context.session.rollback() raise e @mutation def create_contents(self, info: Info, contents: List[CreateContentInput]) -> bool: try: for content in contents: new_content = Content( type=content.type, text=content.text, link=content.link, order=content.order, created_date=datetime.now(), created_by=content.created_by, updated_date=datetime.now(), updated_by=content.created_by ) info.context.session.add(new_content) info.context.session.commit() return True except Exception as e: info.context.session.rollback() raise e @mutation def delete_contents(self, info: Info, contents: List[DeleteContentInput]) -> bool: try: for content in contents: stmt = ( update(Content) .where(Content.id == content.id) .values(is_delete=True) ) result = info.context.session.execute(stmt) info.context.session.commit() return result.rowcount > 0 except Exception as e: info.context.session.rollback() raise e Instead of returning a boolean for the mutation result, please wrap it in a Payload type.
Leave a Comment