30 lines
836 B
Python
30 lines
836 B
Python
"""Fetching file bytes from Telegram's CDN."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
import aiohttp
|
|
|
|
from sticker_downloader.errors import FetchError
|
|
|
|
|
|
@runtime_checkable
|
|
class Fetcher(Protocol):
|
|
"""Minimal interface the downloader needs to read a file off the network."""
|
|
|
|
async def fetch(self, url: str) -> bytes: ...
|
|
|
|
|
|
class AiohttpFetcher:
|
|
"""A :class:`Fetcher` backed by a shared :class:`aiohttp.ClientSession`."""
|
|
|
|
def __init__(self, session: aiohttp.ClientSession) -> None:
|
|
self._session = session
|
|
|
|
async def fetch(self, url: str) -> bytes:
|
|
async with self._session.get(url) as response:
|
|
if response.status >= 400:
|
|
raise FetchError(url, response.status, response.reason or "")
|
|
return await response.read()
|