Untitled

mail@pastecode.io avatar
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