====== Python 3 ====== ===== New features ===== ==== Format string ====


a = 3
b = 'hola'
c = f'{hola} {a} veces'

==== Template strings ====


import os
from string import Template
class TemplatesMixin:
    TEMPLATES_DIR = None
    def _read_template(self, template_path):
        with open(os.path.join(self.TEMPLATES_DIR, template_path)) as template:
            return template.read()
    def render(self, template_path, **kwargs):
        return Template(
            self._read_template(template_path)
        ).substitute(**kwargs)





    Hi $name

==== Merge dicts ====


>>> x = {'a': 1, 'b': 2}
>>> y = {'b': 3, 'c': 4}
>>> z = {**x, **y}
>>> z
{'c': 4, 'a': 1, 'b': 3}

==== Append lists ====


>>> a = [1, 2]
>>> b = [3, 4]
>>> a + b
[1, 2, 3, 4]

==== Data Classes ====


@dataclass
class Process:
    PID: int
    PPID: int
    cmd: str

The ''__init__'' method will already be in your class. Note that here type hinting is required, that is why I have used int and str in the example. If you don't know the type of your field, you [[https://docs.python.org/3/library/typing.html#typing.Any|can use Any from the typing module]]. The Data Class has many advantages compared to the proposed solutions: * It is explicit: all fields are visible, which respects the Zen of Python and makes it readable and maintainable. Compare it to the use of **kwargs and loop over it calling setattr. * It can have methods. Just like any other class. The absence of methods, if you want to use them, is one downside of namedtuple. * It allows you to go beyond the automatic __init__ using the __post_init__ method. ==== Function Annotations ==== You can add documentation to the parameters and the return value:


def add(a:"first number" = 0,
        b:"second number" = 0) -> "sum of a and b":
    return a+b

for item in add.__annotations__.items():
    print(item)

# ('a', 'first number')
# ('b', 'second number')
# ('return', 'sum of a and b')

===== asyncio package ===== Since Python 3.5 we can write co-routines with async/await keywords. Actions on co-routines are allowed by the ''asyncio'' package. Co-routines are functions that in a way similar to green threads are executed in a single thread and process. However, they are not so costly as are not managed by the operating system but by the **event loop**. For an advance use go to [[wiki2:python:notes#advanced_asyncio|Advanced asyncio]]. ==== Basic points ==== * We define co-routines with the ''async'' keyword before a function definition. * We can call co-routines once the event loop has been started. * ''await'' keyword means that at that point is safe to change co-routine. * In other words, the code of a co-routine is synchronous until it finds an ''await''. * Awaitable objects are those that can be used with an ''await'' expression. There are: co-routines, tasks, and futures. === async functions are co-routines ===


async def a_greet(name):
    print (f'Hello {name}')

print(a_greet('Potato'))  #

=== We execute co-routines on the event loop ===


import asyncio

async def a_greet(name):
    print (f'Hello {name}')

asyncio.run(a_greet('Potato'))  # Hello Potato

=== A basic async program ===


import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print(f"started at {time.strftime('%X')}")
    await say_after(1, f"{time.strftime('%X')}: hello")
    await say_after(2, f"{time.strftime('%X')}: world")
    print(f"finished at {time.strftime('%X')}")

asyncio.run(main())
# started at 11:31:12
# 11:31:12: hello
# 11:31:13: world
# finished at 11:31:15

=== It's mandatory to call co-routines with await ===


import asyncio

async def nested():
    return 42

async def main():
    nested()  # RuntimeWarning: coroutine 'nested' was never awaited. It was never called.
    print(await nested())  # will print "42".

asyncio.run(main())

=== We can run several co-routines ===


import asyncio

async def nested(value):
    print(value)

async def main():
    tasks = [nested(i) for i in range(10)]
    await asyncio.gather(*tasks)

asyncio.run(main())

=== You can make async for's and with's ===


async def func():
	async with db.connect() as conn:
		async for user in db.query_users():
			print(user.name)

==== Operations on awaitables ==== Cancelling [[https://docs.python.org/3/library/asyncio-task.html#shielding-from-cancellation|Shielding]]: ''asyncio.shield(async_func_call)'': Avoids the ''async_func_call'' to be cancelled. It can happen that the function that calls the shielded function is cancelled, with this the ''async_func_call'' won't stop given that case. Set a time out with ''coroutine asyncio.wait_for''. Get all unfinished tasks with ''asyncio.all_tasks''. ==== Tasks ==== Tasks are used to schedule co-routines. You create tasks with the ''asyncio.create_task'' function.


import asyncio

async def nested(value):
    return value

async def main():
    task = asyncio.create_task(nested(3))
    await task
    print(task.result())

asyncio.run(main())

Interesting methods are: * ''cancel()'' * ''cancelled()'', ''done()'', to check the status * ''add_done_callback'' While a Task is running in the event loop, no other Tasks can run in the same thread. When a Task executes an await expression, the running Task gets suspended, and the event loop executes the next Task. To schedule a callback from a different OS thread, the ''loop.call_soon_threadsafe()''. ==== The event loop ==== You can run co-routines thanks to the event loop. The most basic function for running a co-routine into the event loop is ''asyncio.run''. You can give it a parameter like a ''main'' co-routine/function which will call other co-routines. This function is a high-level and high-level functions are advised to be used. Functions to manage the event loop are: * ''asyncio.get_running_loop()'' which returns the event loop of the current thread. If it does not exist, an exception will be raised. * ''asyncio.get_event_loop()'' which returns the event loop of the current thread. If it does not exist, one will be crated. * ''asyncio.set_event_loop(loop)'' sets ''loop'' as the current event loop for the current thread. * ''asyncio.new_event_loop()'' Methods of an event loop: * ''run_until_complete(future)'' * ''stop()'' * ''is_running()'' * ''close()'' === Scheduling === ''call_soon'' and ''call_soon_threadsafe'' will a sync function at the next iteration of the loop. ''call_soon_threadsafe'' must be used to schedule callbacks from another thread.


import asyncio

def prnt(value):
    print(f"Hola {value}")

async def arange(count):
    for i in range(count):
        yield(i)

async def main():
    asyncio.get_event_loop().call_soon(prnt, "hello")
    prnt("hola")
    async for i in arange(3):
        print(i)
        await asyncio.sleep(0)

asyncio.run(main(), debug=True)
# Hola hola
# 0
# Hola hello
# 1
# 2

''call_later'' and ''call_at'' allow to set when calling a non-async-function scheduled by loop iterations. ''create_task'' and ''create_future'' . There is also the ''set_task_factory'' if you want to create your own task instances. ==== Futures and executors ==== A ''Future'' represents an eventual result of an asynchronous operation.


async def set_after(fut, delay, value):
    await asyncio.sleep(delay)
    fut.set_result(value)

async def main():
    loop = asyncio.get_running_loop()
    fut = loop.create_future()
    loop.create_task(set_after(fut, 1, '... world'))  # We could have just used "asyncio.create_task()".
    print('hello ...')
    print(await fut)  # Wait until *fut* has a result (1 second) and print it.

asyncio.run(main())

An instance of ''Executor'' provides methods to execute calls asynchronously. It should not be used directly, but through its concrete subclasses.


async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(      # 1. Returns a Future.Executor
        None, blocking_io)
    print('default thread pool', result)
    with concurrent.futures.ThreadPoolExecutor() as pool:      # 2. Run in a custom thread pool: 
        result = await loop.run_in_executor(
            pool, blocking_io)
        print('custom thread pool', result)
    with concurrent.futures.ProcessPoolExecutor() as pool:    # 3. Run in a custom process pool:
        result = await loop.run_in_executor(
            pool, cpu_bound)
        print('custom process pool', result)

''ThreadPoolExecutor'' uses a pool of threads in the same way than ''ProcessPoolExecutor'' uses a pool of processes.


executor = ThreadPoolExecutor(max_workers=2)
a = executor.submit(wait_on_b)
b = executor.submit(wait_on_a)

==== Other tools ==== === Streams === * [[https://docs.python.org/3/library/asyncio-stream.html]] These allow to manage IO resources (sockets) asynchronously. === Synchronization === * [[https://docs.python.org/3/library/asyncio-sync.html]] You can make use of the usual tools for synchronize co-routines (Lock, Event, Condition, Semaphore, BoundedSemaphore). === Queues === * [[https://docs.python.org/3/library/asyncio-queue.html]] Asynchronous queues that can be used to distribute workload between several concurrent tasks:


import asyncio
import random
import time

async def worker(name, queue):
    while True:
        sleep_for = await queue.get()
        await asyncio.sleep(sleep_for)
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    total_sleep_time = 0
    for _ in range(20):
        sleep_for = random.uniform(0.05, 1.0)
        total_sleep_time += sleep_for
        queue.put_nowait(sleep_for)

    # Create three worker tasks to process the queue concurrently.
    tasks = []
    for i in range(3):
        task = asyncio.create_task(worker(f'worker-{i}', queue))
        tasks.append(task)
    # Wait until the queue is fully processed.
    started_at = time.monotonic()
    await queue.join()
    total_slept_for = time.monotonic() - started_at
    # Cancel our worker tasks.
    for task in tasks:
        task.cancel()
    # Wait until all worker tasks are cancelled.
    await asyncio.gather(*tasks, return_exceptions=True)

    print('====')
    print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds')
    print(f'total expected sleep time: {total_sleep_time:.2f} seconds')
asyncio.run(main())

==== Notes on co-routines ==== === Old-school co-routines === These are based on generators (since Python 3.5 they are not useful anymore):


@asyncio.coroutine
def old_style_coroutine():
    yield from asyncio.sleep(1)

async def main():
    await old_style_coroutine()

=== Debug mode === You can debug co-routines using the next configurations: * Setting the PYTHONASYNCIODEBUG environment variable to 1. * Using the -X dev Python command line option. * Passing ''debug=True'' to ''asyncio.run()''. * Calling ''loop.set_debug()''. With this: * asyncio checks for co-routines that were not awaited and logs them * Non-threadsafe asyncio APIs raise an exception if they are called from a wrong thread. * The execution time of the I/O selector is logged if it takes too long to perform an I/O operation. * Callbacks taking longer than 100ms are logged. === Co-routines do not make code fast === * [[https://news.ycombinator.com/item?id=23498742|Discussion on Hacker News]] === functools.partial for passing arguments === ''functools.partial'' allows to create a function that not requires arguments from one that requires them.


loop.call_soon(partial(print, "Hello", flush=True))

=== Stop co-routines when signals ===


import asyncio
import functools
import os
import signal

def ask_exit(signame, loop):
    print("got signal %s: exit" % signame)
    loop.stop()

async def main():
    loop = asyncio.get_running_loop()

    for signame in {'SIGINT', 'SIGTERM'}:
        loop.add_signal_handler(
            getattr(signal, signame),
            functools.partial(ask_exit, signame, loop))

    await asyncio.sleep(3600)

print("Event loop running for 1 hour, press Ctrl+C to interrupt.")
print(f"pid {os.getpid()}: send SIGINT or SIGTERM to exit.")

asyncio.run(main())

===== Enums =====


from enum import Enum

class Numbers(Enum):
    ONE = 1
    TWO = 2

number = Numbers(2)  # 
number1 = Numbers["ONE"]  # 
members = [n for n in Numbers]  # [, ]
values = [n.value for n in Numbers]  # [1, 2]

==== Links ==== * https://realpython.com/python-enum/ ===== Python type checking =====


pi: float = 3.142

def headline(text: str, align: bool = True) -> str:
    pass

The ''MyPy'' library allows to check code files.


>>> names: list = ["Guido", "Jukka", "Ivan"]
>>> version: tuple = (3, 7, 1)
>>> options: dict = {"centered": False, "capitalize": True}
>>> from typing import Dict, List, Tuple
>>> names: List[str] = ["Guido", "Jukka", "Ivan"]
>>> version: Tuple[int, int, int] = (3, 7, 1)
>>> options: Dict[str, bool] = {"centered": False, "capitalize": True}


import random
from typing import Any, Sequence

def choose(items: Sequence[Any]) -> Any:
    return random.choice(items)


class Animal:
    def __init__(self, name: str, birthday: date) -> None:
        self.name = name
        self.birthday = birthday

    @classmethod
    def newborn(cls, name: str) -> "Animal":
        return cls(name, date.today())

    def twin(self, name: str) -> "Animal":
        cls = self.__class__
        return cls(name, self.birthday)


def headline(text, width=80, fill_char="-"):
    # type: (str, int, str) -> str
    return f" {text.title()} ".center(width, fill_char)

def headline(
    text,           # type: str
    width=80,       # type: int
    fill_char="-",  # type: str
):                  # type: (...) -> str
    return f" {text.title()} ".center(width, fill_char)

===== Notes ===== ==== venv para crear entornos virtuales ====


$ python3 -m venv myenv
$ source myenv/bin/activate

To deactivate:


$ deactivate

==== Configure Python3 for a Python2 system ====


# Replace python2 by python3
sudo ln -s /usr/bin/python3.5 /usr/bin/python
# Fix pip3
sudo apt-get remove python3-pip; sudo apt-get install python3-pip
# Install VirtualEnvWrapper for python3
sudo pip3 install virtualenvwrapper

You can force reinstall pip:


curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py
python3 get-pip.py --force-reinstall


sudo python3 -m pip uninstall pip && sudo apt install python3-pip --reinstall
sudo python -m pip uninstall pip && sudo apt install python-pip --reinstall

==== The __all__ keyword ==== It is a list of strings defining what symbols in a module will be exported when ''from import *'' is used on the module.


__all__ = ['bar', 'baz']
waz = 5
bar = 10
def baz(): return 'baz'

These symbols can then be imported like so:


from foo import *
print bar
print baz
# The following will trigger an exception, as "waz" is not exported by the module
print waz

If the ''__all__'' above is commented out, this code will then execute to completion, as the default behavior of ''import *'' is to import all symbols that do not begin with an underscore, from the given namespace. ''__all__'' affects the ''from import *'' behavior only. Members that are not mentioned in ''__all__'' are still accessible from outside the module and can be imported with ''from import '' ==== Gotchas with PIP ==== To install from downloaded packages:


pip install --no-index --find-links=/tmp/python -r requirements.txt