matrix-appservice-kakaotalk/matrix_appservice_kakaotalk/kt/types/request.py

# matrix-appservice-kakaotalk - A Matrix-KakaoTalk puppeting bridge.
# Copyright (C) 2022 Tulir Asokan, Andrew Ferrazzutti
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
from typing import Generic, Type, TypeVar, Union

from attr import dataclass
from enum import IntEnum

from mautrix.types import Serializable, SerializableAttrs, JSON

from .api.auth_api_client import KnownAuthStatusCode


class KnownDataStatusCode(IntEnum):
    SUCCESS = 0
    INVALID_USER = -1
    CLIENT_ERROR = -200
    NOT_LOGON = -201
    INVALID_METHOD = -202
    INVALID_PARAMETER = -203
    INVALID_BODY = -203
    INVALID_HEADER = -204
    UNAUTHORIZED_CHAT_DELETE = -210
    MEDIA_SERVER_ERROR = -300
    CHAT_SPAM_LIMIT = -303
    RESTRICTED_APP = -304
    LOGINLIST_CHATLIST_FAILED = -305
    MEDIA_NOT_FOUND = -306
    MEDIA_THUMB_GEN_FAILED = -307
    UNSUPPORTED = -308
    PARTIAL = -310
    LINK_JOIN_TPS_EXCEEDED = -312
    CHAT_SEND_RESTRICTED = -321
    CHANNEL_CREATE_TEMP_RESTRICTED = -322
    CHANNEL_CREATE_RESTRICTED = -323
    OPENLINK_UNAVAILABLE = -324
    INVITE_COUNT_LIMITED = -325
    OPENLINK_CREATE_RESTRICTED = -326
    INVALID_CHANNEL = -401
    CHAT_BLOCKED_BY_FRIEND = -402
    NOT_CHATABLE_USER = -403
    GAME_MESSAGE_BLOCKED = -406
    BLOCKED_IP = -444
    BACKGROUND_LOGIN_BLOCKED = -445
    OPERATION_DENIED = -500
    CHANNEL_USER_LIMITED = -501
    TEMP_RESTRICTED = -805
    WRITE_WHILE_BLOCKED = -814
    OPENCHAT_REJOIN_REQUIRED = -815
    OPENCHAT_TIME_RESTRICTED = -819
    INVALID_ACCESS_TOKEN = -950
    BLOCKED_ACCOUNT = -997
    AUTH_REQUIRED = -998
    UPDATE_REQUIRED = -999
    SERVER_UNDER_MAINTENANCE = -9797

DataStatusCode = Union[KnownDataStatusCode, int]


@dataclass
class ResponseState(SerializableAttrs):
    status: Union[DataStatusCode, KnownAuthStatusCode] # NOTE Added KnownAuthStatusCode


@dataclass
class RootCommandResult(ResponseState):
    """For brevity, this also encompasses CommandResultFailed and CommandResultDoneVoid"""
    success: bool

    @classmethod
    def deserialize(cls, data: JSON) -> "RootCommandResult":
        if not data or "success" not in data or "status" not in data:
            return RootCommandResult(
                success=True,
                status=KnownDataStatusCode.SUCCESS
            )
        else:
            return super().deserialize(data)


ResultType = TypeVar("ResultType", bound=Serializable)

def ResultListType(result_type: Type[ResultType]):
    """Custom type for setting a result to a list of serializable objects."""
    class _ResultListType(list[result_type], Serializable):
        def serialize(self) -> list[JSON]:
            return [v.serialize() for v in self]

        @classmethod
        def deserialize(cls, data: list[JSON]) -> "_ResultListType":
            return [result_type.deserialize(item) for item in data]

    return _ResultListType

def ResultRawType(result_type: Type):
    """Custom type for setting a result to a primitive."""
    class _ResultRawType(result_type, Serializable):
        def serialize(self) -> result_type:
            return self

        @classmethod
        def deserialize(cls, data: JSON) -> "_ResultRawType":
            return result_type(data)

    return _ResultRawType


@dataclass
class CommandResultDoneValue(RootCommandResult, Generic[ResultType]):
    result: ResultType

    @classmethod
    def deserialize_result(
        cls, result_type: Type[ResultType], data: JSON
    ) -> "CommandResultDoneValue[ResultType]":
        data["result"] = result_type.deserialize(data["result"])
        return cls.deserialize(data)


def deserialize_result(
    result_type: Type[ResultType], data: JSON
) -> Union[CommandResultDoneValue[ResultType], RootCommandResult]:
    """Returns equivalent of CommandResult<T>. Does no consistency checking on success & result properties."""
    if "result" in data and data.get("success"):
        # TODO Handle result object in unsuccessful response
        # TODO Allow arbitrary result object?
        return CommandResultDoneValue.deserialize_result(result_type, data)
    else:
        return RootCommandResult.deserialize(data)


__all__ = [
    "KnownDataStatusCode",
    "DataStatusCode",
    "ResponseState",
    "RootCommandResult",
    "CommandResultDoneValue",
]