Żądania asynchroniczne z żądaniami Pythona

async jest teraz niezależnym modułem: grequests.

Zobacz tutaj: https://github.com/kennethreitz/grequests

I tam: idealna metoda do wysyłania wielu żądań HTTP przez Pythona?

Instalacja:

$ pip install grequests

Użycie:

Zbuduj stos:

import grequests

urls = [
    'http://www.heroku.com',
    'http://tablib.org',
    'http://httpbin.org',
    'http://python-requests.org',
    'http://kennethreitz.com'
]

rs = (grequests.get(u) for u in urls)

Wyślij stos

grequests.map(rs)

Wynik wygląda jak

[<Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>]

Grequesty nie wydają się ustawiać ograniczenia dla jednoczesnych żądań, tj. gdy wiele żądań jest wysyłanych do ten sam serwer.

Przetestowałem zarówno requests-futures jak igrequests . Grequests jest szybszy, ale przynosi łatanie małp i dodatkowe problemy z zależnościami. requests-futures jest kilka razy wolniejszy niż grequests. Postanowiłem napisać własne i po prostu zawinięte żądania do ThreadPoolExecutor i było prawie tak szybkie jak grequesty, ale bez zewnętrznych zależności.

import requests
import concurrent.futures

def get_urls():
    return ["url1","url2"]

def load_url(url, timeout):
    return requests.get(url, timeout = timeout)

with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:

    future_to_url = {executor.submit(load_url, url, 10): url for url in     get_urls()}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            resp_err = resp_err + 1
        else:
            resp_ok = resp_ok + 1

Może requests-futures to inny wybór.

from requests_futures.sessions import FuturesSession

session = FuturesSession()
# first request is started in background
future_one = session.get('http://httpbin.org/get')
# second requests is started immediately
future_two = session.get('http://httpbin.org/get?foo=bar')
# wait for the first request to complete, if it hasn't already
response_one = future_one.result()
print('response one status: {0}'.format(response_one.status_code))
print(response_one.content)
# wait for the second request to complete, if it hasn't already
response_two = future_two.result()
print('response two status: {0}'.format(response_two.status_code))
print(response_two.content)

Jest również zalecany w dokumencie biurowym . Jeśli nie chcesz angażować geventa, to dobrze.

Mam wiele problemów z większością opublikowanych odpowiedzi - albo używają przestarzałych bibliotek, które zostały przeniesione z ograniczonymi funkcjami, albo dostarczają rozwiązanie z zbyt dużą ilością magii na wykonanie żądania, co utrudnia obsługę błędów. Jeśli nie należą do jednej z powyższych kategorii, są bibliotekami innych firm lub są przestarzałe.

Niektóre rozwiązania działają w porządku wyłącznie w żądaniach http, ale rozwiązania nie pasują do każdego innego rodzaju żądania, które to niedorzeczne. Wysoce spersonalizowane rozwiązanie nie jest tutaj konieczne.

Samo użycie wbudowanej biblioteki Pythona asyncio jest wystarczające do wykonywania asynchronicznych żądań dowolnego typu, a także zapewnia wystarczającą płynność dla skomplikowanej i specyficznej obsługi błędów.

import asyncio

loop = asyncio.get_event_loop()

def do_thing(params):
    async def get_rpc_info_and_do_chores(id):
        # do things
        response = perform_grpc_call(id)
        do_chores(response)

    async def get_httpapi_info_and_do_chores(id):
        # do things
        response = requests.get(URL)
        do_chores(response)

    async_tasks = []
    for element in list(params.list_of_things):
       async_tasks.append(loop.create_task(get_chan_info_and_do_chores(id)))
       async_tasks.append(loop.create_task(get_httpapi_info_and_do_chores(ch_id)))

    loop.run_until_complete(asyncio.gather(*async_tasks))

Wiem, że to było zamknięte przez jakiś czas, ale pomyślałem, że może być przydatne promowanie innego rozwiązania asynchronicznego zbudowanego na bibliotece requests.

list_of_requests = ['http://moop.com', 'http://doop.com', ...]

from simple_requests import Requests
for response in Requests().swarm(list_of_requests):
    print response.content

Dokumenty są tutaj: http://pythonhosted.org/simple-requests/

Jeśli chcesz użyć asyncio, to requests-async udostępnia funkcjonalność async / wait dla requests - https://github.com/encode/requests-async

Możesz użyć httpx do tego.

import httpx

async def get_async(url):
    async with httpx.AsyncClient() as client:
        return await client.get(url)

urls = ["http://google.com", "http://wikipedia.org"]

# Note that you need an async context to use `await`.
await asyncio.gather(*map(get_async, urls))

Jeśli chcesz mieć składnię funkcjonalną, gamla lib zawija to w get_async.

Wtedy możesz zrobić

await gamla.map(gamla.get_async(10))(["http://google.com", "http://wikipedia.org"])

10 to limit czasu w sekundach.

(disclaimer: I am its author)

Niestety, z tego co wiem, biblioteka requests nie jest przystosowana do wykonywania żądań asynchronicznych. Możesz zawinąć składnię async/await wokół requests, ale to sprawi, że podstawowe żądania nie będą mniej synchroniczne. Jeśli chcesz uzyskać prawdziwe żądania asynchroniczne, musisz użyć innych narzędzi, które je zapewniają. Jednym z takich rozwiązań jest aiohttp (Python 3.5.3+). Z mojego doświadczenia wynika, że używa się go ze składnią Pythona 3.7 async/await. Poniżej piszę trzy implementacje wykonywania N żądań internetowych using

1. czysto synchroniczne żądania (sync_requests_get_all) za pomocą biblioteki Pythona requests
2. synchroniczne żądania (async_requests_get_all) za pomocą biblioteki Pythona requests owiniętej w składnię Pythona 3.7 async/await i asyncio
3. prawdziwie asynchroniczna implementacja (async_aiohttp_get_all) z biblioteką Pythona aiohttp owiniętą w składnię Pythona 3.7 async/await i asyncio

import time
import asyncio
import requests
import aiohttp

from types import SimpleNamespace

durations = []


def timed(func):
    """
    records approximate durations of function calls
    """
    def wrapper(*args, **kwargs):
        start = time.time()
        print(f'{func.__name__:<30} started')
        result = func(*args, **kwargs)
        duration = f'{func.__name__:<30} finished in {time.time() - start:.2f} seconds'
        print(duration)
        durations.append(duration)
        return result
    return wrapper


async def fetch(url, session):
    """
    asynchronous get request
    """
    async with session.get(url) as response:
        response_json = await response.json()
        return SimpleNamespace(**response_json)


async def fetch_many(loop, urls):
    """
    many asynchronous get requests, gathered
    """
    async with aiohttp.ClientSession() as session:
        tasks = [loop.create_task(fetch(url, session)) for url in urls]
        return await asyncio.gather(*tasks)

@timed
def sync_requests_get_all(urls):
    """
    performs synchronous get requests
    """
    # use session to reduce network overhead
    session = requests.Session()
    return [SimpleNamespace(**session.get(url).json()) for url in urls]


@timed
def async_requests_get_all(urls):
    """
    asynchronous wrapper around synchronous requests
    """
    loop = asyncio.get_event_loop()
    # use session to reduce network overhead
    session = requests.Session()

    async def async_get(url):
        return session.get(url)

    async_tasks = [loop.create_task(async_get(url)) for url in urls]
    return loop.run_until_complete(asyncio.gather(*async_tasks))


@timed
def asnyc_aiohttp_get_all(urls):
    """
    performs asynchronous get requests
    """
    loop = asyncio.get_event_loop()
    return loop.run_until_complete(fetch_many(loop, urls))


if __name__ == '__main__':
    # this endpoint takes ~3 seconds to respond,
    # so a purely synchronous implementation should take
    # little more than 30 seconds and a purely asynchronous
    # implementation should take little more than 3 seconds.
    urls = ['https://postman-echo.com/delay/3']*10

    sync_requests_get_all(urls)
    async_requests_get_all(urls)
    asnyc_aiohttp_get_all(urls)
    print('----------------------')
    [print(duration) for duration in durations]

Na mojej maszynie jest to wyjście:

sync_requests_get_all          started
sync_requests_get_all          finished in 30.92 seconds
async_requests_get_all         started
async_requests_get_all         finished in 30.87 seconds
asnyc_aiohttp_get_all          started
asnyc_aiohttp_get_all          finished in 3.22 seconds
----------------------
sync_requests_get_all          finished in 30.92 seconds
async_requests_get_all         finished in 30.87 seconds
asnyc_aiohttp_get_all          finished in 3.22 seconds

from threading import Thread

threads=list()

for requestURI in requests:
    t = Thread(target=self.openURL, args=(requestURI,))
    t.start()
    threads.append(t)

for thread in threads:
    thread.join()

...

def openURL(self, requestURI):
    o = urllib2.urlopen(requestURI, timeout = 600)
    o...

Od jakiegoś czasu używam zapytań Pythona do asynchronicznych wywołań w GIST API Githuba.

Aby zobaczyć przykład, zobacz kod tutaj:

Https://github.com/davidthewatson/flasgist/blob/master/views.py#L60-72

Ten styl Pythona może nie jest najlepszym przykładem, ale mogę cię zapewnić, że kod działa. Daj mi znać, jeśli to będzie dla Ciebie mylące, a ja to udokumentuję.

Following code creates different threads for each function.

Może to być przydatne w niektórych przypadkach, ponieważ jest prostsze w użyciu. Ale wiedz, że nie jest to asynchroniczne, ale daje iluzję asynchronicznego korzystania z wielu wątków, mimo że dekorator sugeruje, że.

Możesz użyć poniższego dekoratora, aby wywołać wywołanie zwrotne po zakończeniu wykonywania funkcji, wywołanie zwrotne musi obsługiwać przetwarzanie danych zwracanych przez funkcję.

Należy pamiętać, że po udekorowaniu funkcji zwróci ona Future obiekt.

import asyncio

## Decorator implementation of async runner !!
def run_async(callback, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    def inner(func):
        def wrapper(*args, **kwargs):
            def __exec():
                out = func(*args, **kwargs)
                callback(out)
                return out

            return loop.run_in_executor(None, __exec)

        return wrapper

    return inner

Przykład realizacji:

urls = ["https://google.com", "https://facebook.com", "https://apple.com", "https://netflix.com"]
loaded_urls = []  # OPTIONAL, used for showing realtime, which urls are loaded !!


def _callback(resp):
    print(resp.url)
    print(resp)
    loaded_urls.append((resp.url, resp))  # OPTIONAL, used for showing realtime, which urls are loaded !!


# Must provide a callback function, callback func will be executed after the func completes execution
# Callback function will accept the value returned by the function.
@run_async(_callback)
def get(url):
    return requests.get(url)


for url in urls:
    get(url)

Jeśli chcesz zobaczyć, który adres URL jest ładowany w czasie rzeczywistym, możesz dodać następujący kod na końcu, jak również:

while True:
    print(loaded_urls)
    if len(loaded_urls) == len(urls):
        break

Próbowałem też kilku rzeczy używając metod asynchronicznych w Pythonie, jak kiedykolwiek miałem dużo więcej szczęścia używając twisted do programowania asynchronicznego. Ma mniej problemów i jest dobrze udokumentowany. Oto link do tego co próbujesz w twisted.

Http://pythonquirks.blogspot.com/2011/04/twisted-asynchronous-http-request.html