PROJET AUTOBLOG

presque

import os
1 + 1
squares = []
for x in numbers:
    squares.append(x * x)
 
def mean(x):
    return x * x / 2
 
if is_ok:
    print(welcome)
else:
    print(error)
[x * x for x in number]
 
lambda x: x * x / 2
 
print(welcome if is_ok else error)
a = 1
if a = 1:
if (value = Settings.get('test_setting'))
  perform_action(value)
end
value = Settings.get('test_setting')
if value:
  perform_action(value)
while reponse == "oui":
    ...
    reponse = input('Voulez-vous continuer ?')
while reponse = "oui": # erreur subtile et difficile à voir
    ...
    reponse = input('Voulez-vous continuer ?')
match1 = pattern1.match(data)
if match1:
    print(match1.group(1))
else:
    match2 = pattern2.match(data)
    if match2:
        print(match2.group(2))
if (match1 := pattern1.match(data)):
    print(match1.group(1))
elif (match2 := pattern2.match(data)):
    print(match2.group(2))
a = 1
a := 1
(a := 1)
(carottes : Union[Legume, SexToy] = buy())
while True:
    old = total
    total += term
    if old == total:
        return total
    term *= mx2 / (i*(i+1))
    i += 2
while total != (total := total + term):
    term *= mx2 / (i*(i+1))
    i += 2
return total
while (bytes := io.get(x)):
[bar(x) for z in stuff if (x := foo(z))]
diff = x - x_base
if diff:
    g = gcd(diff, n)
    if g > 1:
        return g
if (diff := x - x_base) and (g := gcd(diff, n)) > 1:
    return g
diff = x - x_base
if diff and (g := gcd(diff, n)) > 1:
    return g
class TodoList {
    @observable todos = []; // dire à mobx de surveiller
}
 
...
 
@observer // dire à mobx de tenir à jour
class TodoListView extends Component {
    render() {
        return <ul>
            {this.props.todoList.todos.map(todo =>)}
        </ul>
    }
from dataclasses import dataclass
 
@dataclass
class Achat:
    produit: str
    prix: float
    quantite: int = 0
>>> print(Achat("foo", 2))
Achat(produit='foo', prix=2, quantite=0)
>>> from dataclasses import asdict, astuple
>>> print(asdict(Achat("foo", 2)))
{'produit': 'foo', 'prix': 2, 'quantite': 0}
>>> print(astuple(Achat("foo", 2)))
('foo', 2, 0)
>>> from dataclasses import asdict, astuple
>>> a = Achat("foo", 2)
>>> b = replace(a, quantite=3, prix=1)
>>> print(a, id(a))
Achat(produit='foo', prix=2, quantite=0) 140275795603296
>>> print(b, id(b))
Achat(produit='foo', prix=1, quantite=3) 140275775561456
loop = asyncio.get_event_loop() loop.run_until_complete(coro)
asyncio.run(coro)
def listen_and_answer(sock):
    print('start')
    data = sock.read()
    print('working with %s' % data)
    sock.write('ok')
    print('done')
def listen_and_answer(sock):
    print('start')
 
    def on_listen(data):
        print('working with %s' % data)
 
        def on_write(ret):
            print('done')
 
        sock.write('ok', on_write)
 
    sock.read(on_listen)
async def listen_and_answer(sock):
    print('start')
    data = await sock.read()
    print('working with %s' % data)
    await sock.write('ok')
    print('done')
import asyncio
import random
 
async def succeed(client_writer):
    print('Lucky guy...')
    # Googlez "ayncio high water mark" pour comprender pourquoi c'est
    # une idée à la con de ne pas avoir cette methode asynchrone
    client_writer.write(b'Lucky guy...')
 
async def fail(client_writer):
    raise RuntimeError('Tough shit...')
 
async def handle_request_russian_roulette_style(client_reader, client_writer):
    handlers = (
        succeed,
        succeed,
        succeed,
        fail,
    )
    await handlers[random.randint(0, 3)](client_writer)
    client_writer.close()
 
async def start_server():
    server = await asyncio.start_server(
        handle_request_russian_roulette_style,
        host='localhost', port=8080)
    await server.wait_closed()
 
asyncio.get_event_loop().run_until_complete(start_server())
$ python3 russian_roulette_server.py
Lucky guy...
Lucky guy...
Task exception was never retrieved
future: <Task finished coro=<handle_request_russian_roulette_style() done, defined at ex.py:11> exception=RuntimeError('Tough shit...',)>
Traceback (most recent call last):
  File "ex.py", line 18, in handle_request_russian_roulette_style
    await handlers[random.randint(0, 3)](client_writer)
  File "ex.py", line 9, in fail
    raise RuntimeError('Tough shit...')
RuntimeError: Tough shit...
Lucky guy...
Task exception was never retrieved
future: <Task finished coro=<handle_request_russian_roulette_style() done, defined at ex.py:11> exception=RuntimeError('Tough shit...',)>
Traceback (most recent call last):
  File "ex.py", line 18, in handle_request_russian_roulette_style
    await handlers[random.randint(0, 3)](client_writer)
  File "ex.py", line 9, in fail
    raise RuntimeError('Tough shit...')
RuntimeError: Tough shit...
# pip install trio asks beautifulsoup4
import trio
import asks
import bs4
import re
 
 
# Asks est un grosso modo requests en asynchrone, vu qu'il supporte trio et curio
# (une autre lib asynchrone dans le même style), il faut donc lui dire lequel utiliser
asks.init('trio')
 
 
async def recursive_find(url, on_found, depth=0):
    # On fait notre requête HTTP en asynchrone
    rep = await asks.get(url)
    print(f'depth {depth}, try {url}...')
 
    # On retrouve le corps de l'article grace à beautiful soup
    soup = bs4.BeautifulSoup(rep.text, 'html.parser')
    body = soup.find('div', attrs={"id": 'mw-content-text'})
 
    # On cherche notre point Godwin
    if re.search(r'(?i)hitler|nazi|adolf', body.text):
        on_found(url, depth)
 
    else:
        async with trio.open_nursery() as nursery:
            # On retrouve tous les liens de l'article et relance le recherche
            # de manière récursive
            for tag in body.find_all('a'):
                if tag.has_attr('href') and tag.attrs['href'].startswith('/wiki/'):
                    child_link = 'https://en.wikipedia.org' + tag.attrs['href']
                    # On créé une nouvelle coroutine par lien à crawler
                    nursery.start_soon(recursive_find, child_link, limit, on_found, depth+1)
 
 
async def godwin_find(url):
    limit = trio.CapacityLimiter(10)
    results = []
 
    with trio.move_on_after(10) as cancel_scope:
        def on_found(found_url, depth):
            results.append((found_url, depth))
            cancel_scope.cancel()
 
        await recursive_find(url, limit, on_found)
 
    if results:
        found_url, depth = results[0]
        print(f'Found Godwin point in {found_url} (depth: {depth})')
    else:
        print('No point for this article')
 
 
trio.run(godwin_find, 'https://en.wikipedia.org/wiki/My_Little_Pony')
async with trio.open_nursery() as nursery:
    for tag in body.find_all('a'):
        if tag.has_attr('href') and tag.attrs['href'].startswith('/wiki/'):
            child_link = 'https://en.wikipedia.org' + tag.attrs['href']
            nursery.start_soon(recursive_find, child_link, limit, on_found, depth+1)
coroutines = [recursive_find(link) for link in links]
await asyncio.gather(coroutines)
async def work(sleep_time, nursery):
    await trio.sleep(sleep_time)
    print('work done !')
    # Je vous ai dit qu'une nurserie contient automatiquement un cancel scope ?
    nursery.cancel_scope.cancel()
 
async def work_generator(nursery):
    print('bootstrapping...')
    await trio.sleep(1)
    for sleep_time in range(10):
        nursery.start_soon(nursery, sleep_time, nursery)
 
async def stop_a_first_work_done():
    async with trio.open_nursery() as nursery:
        await work_generator(nursery)
        print('Waiting for a work to finish...')
with trio.move_on_after(10) as cancel_scope:
    def on_found(found_url, depth):
        results.append((found_url, depth))
        cancel_scope.cancel()
 
    await recursive_find(url, limit, on_found)
import trio_asyncio
import asyncpg
 
 
class TrioConnProxy:
    # Le décorateur permet de marquer la frontière entre trio et asyncio
    @trio_asyncio.trio2aio
    async def init(self, url):
        # Ici on est donc dans asyncio
        self.conn = await asyncpg.connect(url)
 
    @trio_asyncio.trio2aio
    async def execute(self, *args):
        return await self.conn.execute(*args)
 
    @trio_asyncio.trio2aio
    async def fetch(self, *args):
        return await self.conn.fetch(*args)
 
 
async def main():
    # Ici on est dans trio, c'est la fête
 
    conn = TrioConnProxy()
    await conn.init('postgresql:///')
 
    await conn.execute('CREATE TABLE IF NOT EXISTS users(name text primary key)')
 
    for name in ('Riri', 'Fifi', 'Loulou'):
        await conn.execute('INSERT INTO users(name) VALUES ($1)', name)
 
    users = await conn.fetch('SELECT * FROM users')
    print('users:', [user[0] for user in users])
 
 
# trio_asyncio s'occupe de configurer l'event loop par défaut de asyncio
# puis lance le <code>trio.run</code> classique
trio_asyncio.run(main)

PROJET AUTOBLOG

Sam et Max

Guido van Rossum s’offre “des vacances permanentes” 2

Python-idea, Python-dev et les PEP

Pride and prejudice

Un anus pour les rassembler tôt

So what now ?

Quoi qu’il arrive

L’expression d’assignation vient d’être acceptée 12 Recently updated !

Un peu d’histoire

La PEP 572

Une opinion, c’est comme un trou du cul…

Le don du mois: mobx 2 Recently updated !

Python 3.7 sort de sa coquille 4 Recently updated !

Les data classes

asyncio++

Des perfs

breakpoint()

Quelques détails

Dicts ordonnés

async/await sont des mots clés

Des soeurs pour getitem et getattr

Traduction de la doc

DeprecationWarning plus visible

Debug mode

Des pyc reproductibles

Meilleur ImportError

packaging

time est plus précis

unittest -k

namedtuple supporte les valeurs par defaut

Ajouts à Contexlib

lifting pour subprocess

Super article invité sur Trio que l’auteur a oublié de titrer Recently updated !

1 – Pourquoi c’est (de) la merde ?

2 – Trio, une façon de faire de l’asynchrone

3 – L’écosystème

Python-idea, Python-dev et les PEP

Pride and prejudice

Un anus pour les rassembler tôt

So what now ?

Quoi qu’il arrive

Un peu d’histoire

La PEP 572

Une opinion, c’est comme un trou du cul…

Les data classes

asyncio++

Des perfs

breakpoint()

Quelques détails

Dicts ordonnés

async/await sont des mots clés

Des soeurs pour __getitem__ et __getattr__

Traduction de la doc

DeprecationWarning plus visible

Debug mode

Des pyc reproductibles

Meilleur ImportError

packaging

time est plus précis

unittest -k

namedtuple supporte les valeurs par defaut

Ajouts à Contexlib

lifting pour subprocess

1 – Pourquoi c’est (de) la merde ?

2 – Trio, une façon de faire de l’asynchrone

3 – L’écosystème

Des soeurs pour getitem et getattr