summaryrefslogtreecommitdiffstats
path: root/g4f/requests.py
blob: f6f2383bb398b12bde55bfc99e9345885c1c9b7e (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
from __future__ import annotations

import json, sys, asyncio
from functools import partialmethod

from aiohttp import StreamReader
from aiohttp.base_protocol import BaseProtocol

from curl_cffi.requests import AsyncSession as BaseSession
from curl_cffi.requests import Response
from curl_cffi import AsyncCurl

is_newer_0_5_9 = hasattr(AsyncCurl, "remove_handle")


class StreamResponse:
    def __init__(self, inner: Response, content: StreamReader, request):
        self.inner = inner
        self.content = content
        self.request = request
        self.status_code = inner.status_code
        self.reason = inner.reason
        self.ok = inner.ok
        self.headers = inner.headers
        self.cookies = inner.cookies

    async def text(self) -> str:
        content = await self.content.read()
        return content.decode()

    def raise_for_status(self):
        if not self.ok:
            raise RuntimeError(f"HTTP Error {self.status_code}: {self.reason}")

    async def json(self, **kwargs):
        return json.loads(await self.content.read(), **kwargs)

class StreamRequest:
    def __init__(self, session: AsyncSession, method: str, url: str, **kwargs):
        self.session = session
        self.loop = session.loop if session.loop else asyncio.get_running_loop()
        self.content = StreamReader(
            BaseProtocol(session.loop),
            sys.maxsize,
            loop=session.loop
        )
        self.method = method
        self.url = url
        self.options = kwargs

    def on_content(self, data):
        if not self.enter.done():
            self.enter.set_result(None)
        self.content.feed_data(data)

    def on_done(self, task):
        if not self.enter.done():
            self.enter.set_result(None)
        self.content.feed_eof()

    async def __aenter__(self) -> StreamResponse:
        self.curl = await self.session.pop_curl()
        self.enter = self.loop.create_future()
        request, _, header_buffer = self.session._set_curl_options(
            self.curl,
            self.method,
            self.url,
            content_callback=self.on_content,
            **self.options
        )
        if is_newer_0_5_9:
             self.handle = self.session.acurl.add_handle(self.curl)
        else:
            await self.session.acurl.add_handle(self.curl, False)
            self.handle = self.session.acurl._curl2future[self.curl]
        self.handle.add_done_callback(self.on_done)
        await self.enter
        if is_newer_0_5_9:
            response = self.session._parse_response(self.curl, _, header_buffer)
            response.request = request
        else:
            response = self.session._parse_response(self.curl, request, _, header_buffer)
        return StreamResponse(
            response,
            self.content,
            request
        )
            
    async def __aexit__(self, exc_type, exc, tb):
        if not self.handle.done():
            self.session.acurl.set_result(self.curl)
        self.curl.clean_after_perform()
        self.curl.reset()
        self.session.push_curl(self.curl)   

class AsyncSession(BaseSession):
    def request(
        self,
        method: str,
        url: str,
        **kwargs
    ) -> StreamRequest:
        return StreamRequest(self, method, url, **kwargs)
    
    head = partialmethod(request, "HEAD")
    get = partialmethod(request, "GET")
    post = partialmethod(request, "POST")
    put = partialmethod(request, "PUT")
    patch = partialmethod(request, "PATCH")
    delete = partialmethod(request, "DELETE")