course/w8/w8-s4-av-code.html at 72b3a74b90454429fb6da52a2a58fa4e0224f36c · flotpython/course · GitHub

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
<div class='clipboard'><style>
.clipboard {
    font-size: 11px;
}
</style>
<h3 style="border: 1px solid #66b; border-radius: 5px; padding: 10px; background-color: #eee;">
<b><i>bloc-note pour copier/coller le code de la vidéo</i></b>
</h3>
<hr />
<!-- Version Python 3.10 -->
<h3 style="border: 1px solid #66b; border-radius: 5px; padding: 10px; background-color: #eee;">
<b><i>La vidéo a été tournée avec la 3.6, le cours a été mis à jour pour 3.10.
Il est préférable d'utiliser les fonctions asyncio de haut niveau, telles qu'asyncio.run().
Le référencement de l'objet boucle ou l'appel de ses méthodes s'adresse principalement aux auteurs de code,
de bibliothèques et de frameworks de niveau inférieur, qui ont besoin d'un contrôle plus fin du comportement de la boucle d'événement.
</i></b>
</h3>
<hr />
<blockquote>
<p>Prendre un notebook ou un éditeur python</p>
</blockquote>
<pre><code>import time

urls = [&quot;https://www.irs.gov/pub/irs-pdf/f1040.pdf&quot;,
		&quot;https://www.irs.gov/pub/irs-pdf/f1040es.pdf&quot;,
		&quot;https://www.irs.gov/pub/irs-pdf/f1040c.pdf&quot;,
		&quot;https://www.irs.gov/pub/irs-pdf/f1040esn.pdf&quot;,
		]

import requests

beg = time.time()

for url in urls:
    req = requests.get(url)
    print(f&quot;{url} returned {len(req.text)} chars&quot;)

print(f&quot;duration = {time.time() - beg}s&quot;)
</code></pre>
<blockquote>
<p>Prendre l'éditeur IDLE sur votre machine, après avoir installé aiohttp (pip install aiohttp)</p>
</blockquote>
<pre><code>import asyncio
import aiohttp
import time
import platform

if platform.system()=='Windows':
    """Solve RuntimeError: Event loop is closed in Windows"""
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

urls = [&quot;https://www.irs.gov/pub/irs-pdf/f1040.pdf&quot;,
		&quot;https://www.irs.gov/pub/irs-pdf/f1040es.pdf&quot;,
		&quot;https://www.irs.gov/pub/irs-pdf/f1040c.pdf&quot;,
		&quot;https://www.irs.gov/pub/irs-pdf/f1040esn.pdf&quot;,
		]


async def fetch(url):

    async with aiohttp.ClientSession() as session:
        print(f&quot;fetching {url}&quot;)

        async with session.get(url) as response:
            raw = await response.read()
            print(f&quot;{url} returned {len(raw)} bytes&quot;)

async def fetch_urls():
    await asyncio.gather(*(fetch(url) for url in urls))
</code></pre>
<blockquote>
<p>Faire Ctl-s puis F5</p>
</blockquote>
<pre><code>
beg = time.time()
asyncio.run(fetch_urls())
print(f&quot;duration = {time.time() - beg}s&quot;)
</code></pre>
<blockquote>
<p>Reprendre l'éditeur python sur votre machine et ajouter</p>
</blockquote>
<pre><code>async def fetch2(url, i):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            # avec ici une itération asynchrone
            async for line in response.content:
                print(f&quot;{i}&quot;, end=&quot;&quot;)
    return url

async def run_fetch2():
    await asyncio.gather(*(fetch2(url, i) for i, url in enumerate(urls)))

asyncio.run(run_fetch2())
</code></pre>
</div>