Autoblog de sametmax.com

presque

    assert condition

    assert condition, "Message en cas d'erreur"

    ARTIBUSES = {
        'truc': 1,
        'bidule': 1,
        'machin': '2',
        'chose': '3',
        'chouette': '3',
        'foo': '4',
        # ...
    }
    
    def souquer_les_artibuses(artibuse):
    
        target = ARTIBUSES[artibuse]
        # un code de souquage professionnel s'ensuit

    def souquer_les_artibuses(artibuse):
        if artibuse not in ARTIBUSES:
            raise ValueError(
                f"'{artibuse}'' n'est pas une artibuse valide. "
                f"Utilisez une valeur parmi: {', '.join(ARTIBUSES)}"
            )
        target = ARTIBUSES[artibuse]


    def souquer_les_artibuses(artibuse):

        assert artibuse in ARTIBUSES, (
            f"'{artibuse}'' n'est pas une artibuse valide. "
            f"Utilisez une valeur parmi: {', '.join(ARTIBUSES)}}"
        )

        target = ARTIBUSES[artibuse]

if __debug__:
    un_dump_log_de_porc()

>>> class = 1
  File "", line 1
    class = 1
          ^
SyntaxError: invalid syntax
>>> class Foo: pass
... 
>>> Foo.class = 1
  File "", line 1
    Foo.class = 1
            ^
SyntaxError: invalid syntax
>>> 
>>> class Bar:
...     @classmethod
...     def wololo(cls):
...         print(cls, 'wololo')
... 
>>> 
>>> Bar.wololo()
 wololo
>>> 
python -Wd
>>> def async():
...     pass
... 
:1: DeprecationWarning: 'async' and 'await' will become reserved keywords in Python 3.7
python -Wa
$ python3.6
Python 3.6.5 (default, May  3 2018, 10:08:28) 
[GCC 5.4.0 20160609] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import imp
$ python3.7 
Python 3.7.0+ (default, Jun 28 2018, 14:08:14) 
[GCC 5.4.0 20160609] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import imp
__main__:1: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses

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)

squares = [x * x for x in number]

mean = 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)
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 (data := 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 
            {this.props.todoList.todos.map(todo =>)}
        
    }


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 replace
>>> 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:  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:  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, on_found, depth+1)


async def godwin_find(url):
    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, 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, 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(work, 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, 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 trio.run classique trio_asyncio.run(main)

Autoblog de sametmax.com

Arrive setup.cfg

Setup.cfg, ou es-tu ?

pyproject.toml vient foutre le bordel

N’est-ce pas trop tard ?

Effets secondaires de -o

Le problème avec assert

Limites et contournements des mots clés

Arrive la 3.7

Comment on aurait pu éviter ce merdier ?

La programmation, c’est BEAUCOUP de boulot

Aides techniques

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 sœurs pour __getitem__ et __getattr__

Traduction de la doc

DeprecationWarning plus visibles

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 – Eeeeet c’est tout !

4 – L’écosystème

Des sœurs pour getitem et getattr