Add upload_file method

Author: Palbahngmiyine

solapi/error/MessageNotReceiveError.py

@@ -4,7 +4,7 @@
 class MessageNotReceivedError(Exception):
     def __init__(self, failed_messages: list[FailedMessage]):
         self.failed_messages = failed_messages
-        
+
     def __str__(self):
         # TODO: i18n needed
-        return "메시지 접수에 실패했습니다.\n자세한 내용은 MessageNotReceivedError의 failed_messages 프로퍼티를 참조해주세요."
+        return "메시지 접수에 실패했습니다.\n자세한 내용은 MessageNotReceivedError의 failed_messages 프로퍼티를 참조해 주세요."

solapi/message_service.py

@@ -1,4 +1,6 @@
+import base64
 from datetime import datetime
+from pathlib import Path
 from typing import Optional, Union
 
 from solapi.error.MessageNotReceiveError import MessageNotReceivedError
@@ -10,7 +12,9 @@
     SendMessageRequest,
     SendRequestConfig,
 )
+from solapi.model.request.storage import FileTypeEnum, FileUploadRequest
 from solapi.model.response.send_message_response import SendMessageResponse
+from solapi.model.response.storage import FileUploadResponse
 
 
 class SolapiMessageService:
@@ -77,3 +81,27 @@ def send(
             raise MessageNotReceivedError(failed_messages) from ValueError
 
         return deserialized_response
+
+    def upload_file(
+        self, file_path: str, upload_type: FileTypeEnum = FileTypeEnum.MMS
+    ) -> FileUploadResponse:
+        path = Path(file_path)
+        with open(path, "rb") as image_file:
+            encoded_string = base64.b64encode(image_file.read())
+
+        request = FileUploadRequest(
+            file=str(encoded_string)[2:-1],
+            type=upload_type,
+        ).model_dump(exclude_none=True)
+        response = default_fetcher(
+            self.auth_info,
+            request={
+                "url": f"{self.base_url}/storage/v1/files",
+                "method": RequestMethod.POST,
+            },
+            data=request,
+        )
+        deserialized_response: FileUploadResponse = FileUploadResponse.model_validate(
+            response
+        )
+        return deserialized_response

solapi/model/message.py

@@ -100,5 +100,7 @@ class Message(BaseModel):
     )
 
     model_config = ConfigDict(
-        extra="ignore", populate_by_name=True, alias_generator=to_camel
+        extra="ignore",
+        populate_by_name=True,
+        alias_generator=to_camel,
     )

solapi/model/request/storage.py

@@ -0,0 +1,18 @@
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class FileTypeEnum(str, Enum):
+    MMS = "MMS"
+    KAKAO = "KAKAO"
+    RCS = "RCS"
+    FAX = "FAX"
+
+
+class FileUploadRequest(BaseModel):
+    type: FileTypeEnum
+    file: str
+    name: Optional[str] = None
+    link: Optional[str] = None

solapi/model/response/send_message_response.py

@@ -14,9 +14,12 @@ class MessageItem(BaseModel):
 
     model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
 
+
 class FailedMessage(BaseModel):
     to: Optional[str]
-    from_: Optional[str] = Field(..., serialization_alias="from", validation_alias="from")
+    from_: Optional[str] = Field(
+        ..., serialization_alias="from", validation_alias="from"
+    )
     type: str
     status_message: str = Field(..., alias="statusMessage")
     country: Optional[str]

solapi/model/response/storage.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+
+
+class KakaoFileUploadResponseProperty(BaseModel):
+    daou: Optional[str] = None
+    biztalk: Optional[str] = None
+
+
+class FileUploadResponse(BaseModel):
+    kakao: Optional[KakaoFileUploadResponseProperty] = None
+    type: str
+    original_name: str
+    link: Optional[str] = None
+    file_id: str
+    name: str
+    url: str
+    account_id: str
+    references: Optional[list[str]] = None
+    date_created: Optional[datetime] = None
+    date_updated: Optional[datetime] = None
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )