Strawberry (GraphQL)で特定のエラーのみをエラーログ出力する

PythonのGraphQLライブラリであるStrawberryを使用していて、困ったことがあります。それは、GraphQLの実行エラーがすべてエラーログに吐き出されてしまうことです。その結果、クライアントエラーやハンドリングされていない未知のエラーも含めて、全てのエラーがクラッシュレポートツール（Sentry）に報告されてしまう問題が発生しました。

ここでは、Strawberryを使用して特定のエラーのみをエラーログに出力する方法を示します。

動作を確認した環境

• Python3.9
• Strawberry 0.187

解決方法

解決方法は、CustomSchemaを作成し、process_errorsをオーバーライドし、そこでエラーの種類に応じてエラーログの出力を制御することです。

class CustomSchema(Schema):
    def __init__(self, not_log_target_errors: List[Any], *args, **kwargs):
        self.not_target_errors = not_log_target_errors
        super().__init__(*args, **kwargs)

    def is_log_target_error(self, obj) -> bool:
        return not any(isinstance(obj, not_target_error) for not_target_error in self.not_target_errors)

    def process_errors(self, errors: List[GraphQLError], execution_context: Optional[ExecutionContext] = None):
        log_target_errors = [error for error in errors if self.is_log_target_error(error.original_error)]
        for log_target_error in log_target_errors:
            StrawberryLogger.error(log_target_error, execution_context)


schema = CustomSchema(query=Query, not_log_target_errors=not_log_target_errors_for_app)

以下のコードでunite test済みです。

from textwrap import dedent

import strawberry

import CustomSchema


class TestCustomSchema:
    def _get_schema_and_query(self):
        class TargetError(Exception):
            pass

        class NotTargetError1(Exception):
            pass

        class NotTargetError2(Exception):
            pass

        not_target_errors = [NotTargetError1, NotTargetError2]

        @strawberry.type
        class Query:
            @strawberry.field
            def target_error_endpoint(self) -> str:
                raise TargetError("this is TargetError message")

            @strawberry.field
            def not_target_error_endpoint_1(self) -> str:
                raise NotTargetError1("this is NotTargetError1 message")

            @strawberry.field
            def not_target_error_endpoint_2(self) -> str:
                raise NotTargetError2("this is NotTargetError2 message")

        schema = CustomSchema(query=Query, not_log_target_errors=not_target_errors)
        return schema, Query

    def test_process_errors_target_errors(self, caplog):
        schema, query = self._get_schema_and_query()

        query_target_error_endpoint = dedent(
            """
            query {
                targetErrorEndpoint
            }
        """
        ).strip()

        result = schema.execute_sync(
            query_target_error_endpoint,
            root_value=query(),
        )
        assert len(result.errors) == 1
        assert result.errors[0].message == "this is TargetError message"
        assert len(caplog.records) == 1
        record = caplog.records[0]

        assert record.levelname == "ERROR"
        assert record.name == "strawberry.execution"
        assert "this is TargetError message" in caplog.records[0].message

    def test_process_errors_not_target_errors_1(self, caplog):
        schema, query = self._get_schema_and_query()

        query_not_target_error_endpoint_1 = dedent(
            """
            query {
                notTargetErrorEndpoint1
            }
        """
        ).strip()

        result = schema.execute_sync(
            query_not_target_error_endpoint_1,
            root_value=query(),
        )
        assert len(result.errors) == 1
        assert result.errors[0].message == "this is NotTargetError1 message"
        assert len(caplog.records) == 0

    def test_process_errors_not_target_errors_2(self, caplog):
        schema, query = self._get_schema_and_query()

        query_not_target_error_endpoint_2 = dedent(
            """
            query {
                notTargetErrorEndpoint2
            }
        """
        ).strip()

        result = schema.execute_sync(
            query_not_target_error_endpoint_2,
            root_value=query(),
        )
        assert len(result.errors) == 1
        assert result.errors[0].message == "this is NotTargetError2 message"
        assert len(caplog.records) == 0

参考↓↓
override する前の process_errorsです。
(出典: https://strawberry.rocks/docs/types/schema#handling-execution-errors)
GraphQLError を全てエラーログ出力する実装になっています。

# strawberry/schema/base.py
from strawberry.types import ExecutionContext
 
logger = logging.getLogger("strawberry.execution")




class BaseSchema:
    ...


    def process_errors(
        self,
        errors: List[GraphQLError],
        execution_context: Optional[ExecutionContext] = None,
    ) -> None:
        for error in errors:
            StrawberryLogger.error(error, execution_context)

# strawberry/utils/logging.py
from strawberry.types import ExecutionContext




class StrawberryLogger:
    logger: Final[logging.Logger] = logging.getLogger("strawberry.execution")


    @classmethod
    def error(
        cls,
        error: GraphQLError,
        execution_context: Optional[ExecutionContext] = None,
        # https://www.python.org/dev/peps/pep-0484/#arbitrary-argument-lists-and-default-argument-values
        **logger_kwargs: Any,
    ) -> None:
        # "stack_info" is a boolean; check for None explicitly
        if logger_kwargs.get("stack_info") is None:
            logger_kwargs["stack_info"] = True


        # stacklevel was added in version 3.8
        # https://docs.python.org/3/library/logging.html#logging.Logger.debug
        if sys.version_info >= (3, 8):
            logger_kwargs["stacklevel"] = 3


        cls.logger.error(error, exc_info=error.original_error, **logger_kwargs)

宣伝

最後まで読んで頂きありがとうございました。

先日、 Strawberry GraphQL にコミットできました。
Strawberry GraphQL を使っていく皆様、フォローをお願いします。

PRマージされた🎉
文化圏が全く違う方とのやりとりめっちゃ緊張した。
例外を足しただけだが、良い経験させてもらった。 https://t.co/fiCkyNbkAe

— nassy@PythonとAzure (@n_nassy20) December 2, 2022