Skip to content

Commit

Permalink
Use Bot Access Key if available for post_message_attachment (#67)
Browse files Browse the repository at this point in the history
Co-authored-by: Jelle Zijlstra <[email protected]>
  • Loading branch information
krisyang1125 and JelleZijlstra authored Feb 24, 2024
1 parent 1300974 commit 53f9594
Showing 1 changed file with 54 additions and 2 deletions.
56 changes: 54 additions & 2 deletions src/fastapi_poe/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from sse_starlette.sse import EventSourceResponse, ServerSentEvent
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import Message
from typing_extensions import deprecated, overload

from fastapi_poe.types import (
AttachmentUploadResponse,
Expand Down Expand Up @@ -148,6 +149,12 @@ async def on_error(self, error_request: ReportErrorRequest) -> None:
def __post_init__(self) -> None:
self._pending_file_attachment_tasks = {}

# This overload leaves access_key as the first argument, but is deprecated.
@overload
@deprecated(
"The access_key parameter is deprecated. "
"Set the access_key when creating the Bot object instead."
)
async def post_message_attachment(
self,
access_key: str,
Expand All @@ -158,7 +165,35 @@ async def post_message_attachment(
filename: Optional[str] = None,
content_type: Optional[str] = None,
is_inline: bool = False,
) -> AttachmentUploadResponse: ...

# This overload requires all parameters to be passed as keywords
@overload
async def post_message_attachment(
self,
*,
message_id: Identifier,
download_url: Optional[str] = None,
file_data: Optional[Union[bytes, BinaryIO]] = None,
filename: Optional[str] = None,
content_type: Optional[str] = None,
is_inline: bool = False,
) -> AttachmentUploadResponse: ...

async def post_message_attachment(
self,
access_key: Optional[str] = None,
message_id: Optional[Identifier] = None,
*,
download_url: Optional[str] = None,
file_data: Optional[Union[bytes, BinaryIO]] = None,
filename: Optional[str] = None,
content_type: Optional[str] = None,
is_inline: bool = False,
) -> AttachmentUploadResponse:
if message_id is None:
raise InvalidParameterError("message_id parameter is required")

task = asyncio.create_task(
self._make_file_attachment_request(
access_key=access_key,
Expand All @@ -182,20 +217,37 @@ async def post_message_attachment(

async def _make_file_attachment_request(
self,
access_key: str,
message_id: Identifier,
*,
access_key: Optional[str] = None,
download_url: Optional[str] = None,
file_data: Optional[Union[bytes, BinaryIO]] = None,
filename: Optional[str] = None,
content_type: Optional[str] = None,
is_inline: bool = False,
) -> AttachmentUploadResponse:
if self.access_key:
if access_key:
warnings.warn(
"Bot already has an access key, access_key parameter is not needed.",
DeprecationWarning,
stacklevel=2,
)
attachment_access_key = access_key
else:
attachment_access_key = self.access_key
else:
if access_key is None:
raise InvalidParameterError(
"access_key parameter is required if bot is not"
+ " provided with an access_key when make_app is called."
)
attachment_access_key = access_key
url = "https://www.quora.com/poe_api/file_attachment_3RD_PARTY_POST"

async with httpx.AsyncClient(timeout=120) as client:
try:
headers = {"Authorization": f"{access_key}"}
headers = {"Authorization": f"{attachment_access_key}"}
if download_url:
if file_data or filename:
raise InvalidParameterError(
Expand Down

0 comments on commit 53f9594

Please sign in to comment.