Concurrency

import asyncio

async def main():
  await asyncio.sleep(2)
  print('Hello World!')


asyncio.run(main())

def add_one(number: int) -> int:
  return number + 1

async def coroutine_add_one(number: int) -> int:
  return number + 1

function_result = add_one(1)
coroutine_result = asyncio.run(coroutine_add_one(1))

print(f'Function result is {function_result} and the type is {type(function_result)}')
print(f'Coroutine result is {coroutine_result} and the type is {type(coroutine_result)}')

import asyncio

async def add_one(number: int) -> int:
  return number + 1

async def main() -> None:
  one_plus_one = await add_one(1) # pause and wait for the result of add_one(1)
  two_plus_one = await add_one(2) # pause and wait for the result of add_one(2)

  print(one_plus_one)
  print(two_plus_one)

asyncio.run(main())

import asyncio
from util import delay # our module

async def add_one(number: int) -> int:
  return number + 1

async def hello_world_message() -> str:
  await delay(1)
  return 'Hello world!'

async def main() -> None:
  message = await hello_world_message()
  one_plus_one = await add_one(1) 
  print(one_plus_one)
  print(message)

asyncio.run(main())

from asyncio import Future
import asyncio

def make_request() -> Future:
  future = Future()
  asyncio.create_task(set_future_value(future)) # create a task asynchronously set the value of the future
  return future

async def set_future_value(future) -> None:
  await asyncio.sleep(1) # wait 1 sec before setting the value of future
  future.set_result(42)

async def main():
  future = make_request()
  print(f'Is the future done? {future.done()}')
  value = await future # pause main until the future is set
  print(f'Is the future done? {future.done()}')
  print(value)

asyncio.run(main())

import asyncio
from util import delay

async def main():
  sleep_for_three = asyncio.create_task(delay(3))
  print(type(sleep_for_three))
  result = await sleep_for_three
  print(result)

asyncio.run(main())

import asyncio
from util.delay import delay

async def main():
  sleep_for_three = asyncio.create_task(delay(3))
  sleep_again = asyncio.create_task(delay(3))
  sleep_once_more = asyncio.create_task(delay(3))

  await sleep_for_three
  await sleep_again
  await sleep_once_more

asyncio.run(main())

import asyncio
from util.delay import delay

async def hello_every_second():
  for i in range(2):
    await asyncio.sleep(1)
    print('I am running other code while I am running')

async def main():
  first_delay = asyncio.create_task(delay(3))
  second_delay = asyncio.create_task(delay(3))

  await hello_every_second()
  await first_delay
  await second_delay

asyncio.run(main())

import asyncio
from util.delay import delay
from asyncio import CancelledError

async def main():
  long_task = asyncio.create_task(delay(10))
  seconds_elapsed = 0
  while not long_task.done():
    print('Task is not finished, checking again in a second.')
    await asyncio.sleep(1)
    seconds_elapsed += 1
    if seconds_elapsed == 5:
      long_task.cancel()

  try:
    await long_task
  except CancelledError:
    print('Our task was cancelled.')

asyncio.run(main())

import asyncio
from util.delay import delay

async def main():
  delay_task = asyncio.create_task(delay(2))
  try:
    result = await asyncio.wait_for(delay_task, timeout=1)
    print(result)
  except asyncio.exceptions.TimeoutError:
    print('Got a timeout!')
    print(f'Was the task cancelled? {delay_task.cancelled()}')

asyncio.run(main())

import asyncio
from util.delay import delay

async def main():
  delay_task = asyncio.create_task(delay(10))

  try:
    result = await asyncio.wait_for(asyncio.shield(delay_task), 5)
    print(result)
  except asyncio.exceptions.TimeoutError:
    print('Task took longer than five seconds, it will finish soon')
    result = await delay_task
    print(f'finished in {result} second(s)')

asyncio.run(main())

import asyncio
from aiohttp import ClientSession, ClientTimeout
from util import fetch_status

async def main():
  session_timeout = ClientTimeout(total=1, connect=.1)
  async with ClientSession(timeout=session_timeout) as session:  # creates a session
    await fetch_status(session, 'https://example.com')  # returns a status code for the given URL

asyncio.run(main())

import asyncio
from util import async_timed, delay

@async_timed()
async def main() -> None:
  delay_times = [ 3, 3, 3 ]
  tasks = [ await asyncio.create_task(delay(seconds)) for seconds in delay_times ]
  [ await task for task in tasks ]

asyncio.run(main())

import asyncio
from aiohttp import ClientSession
from util import async_timed, fetch_status

@async_timed()
async def main() -> None:
  async with ClientSession as session:
    urls = [ 'https://example.com' for _ in range(1000) ]
    requests = [ fetch_status(session, url) for url in urls ] # generates a list of coroutines for each request we want to make
    status_codes = await asyncio.gather(*requests) # waits for all requests to complete

asyncio.run(main())

import asyncio
from aiohttp import ClientSession
from util import fetch_status

async def main():
  async with ClientSession as session:
    fetchers = [ fetch_status(session, 'https://www.example.com', 1), # coroutine 1
                 fetch_status(session, 'https://www.example.com', 1), # coroutine 2
                 fetch_status(session, 'https://www.example.com', 10), # coroutine 3
               ]

    for finished_task in asyncio.as_completed(fetchers, timeout=2):  # If it takes longer than the timeout, each awaitable in the iterator will throw a TimeoutException when we await it.

      try:
        result = await done_task
        print(result)
      except asyncio.TimeoutError:
        print('We got a timeout error')

      for task in asyncio.tasks.all_tasks():
        print(task)

asyncio.run(main())

import asyncio
import logging

from aiohttp import ClientSession
from util import fetch_status

async def main() -> None:
  async with ClientSession as session:
    good_request = fetch_status(session, 'https://www.example.com')
    bad_request = fetch_status(session, 'python://bad'))
    fetchers = [ asyncio.create_task(good_request),
                 asyncio.create_task(bad_request)
               ]
  done, pending = await asyncio.wait(fetchers)
  for done_task in done:
    # result = await done_task will throw an exception
    if done_task.exception() is None: # checks to see if we have an exception
      print(done_task.result())
    else:
      logging.error("Request got an exception", exc_info=done_task.exception())

asyncio.run(main())

import asyncio
import logging

from aiohttp import ClientSession
from util import fetch_status

async def main() -> None:
  async with ClientSession as session:
    fetchers = [ asyncio.create_task(fetch_status(session, 'python://bad')),
                 asyncio.create_task(fetch_status(session, 'https://www.example.com', delay=3)),
                 asyncio.create_task(fetch_status(session, 'https://www.example.com', delay=3)),
    ]
  done, pending = await asyncio.wait(fetchers, return_when=asyncio.FIRST_EXCEPTION)
  for done_task in done:
    # result = await done_task will throw an exception
    if done_task.exception() is None: # checks to see if we have an exception
      print(done_task.result())
    else:
      logging.error("Request got an exception", exc_info=done_task.exception())
  for pending_task in pending:
    pending_task.cancel()
asyncio.run(main())

import asyncio
import logging

from aiohttp import ClientSession
from util import fetch_status

async def main() -> None:
  async with ClientSession as session:
    fetchers = [ asyncio.create_task(fetch_status(session, 'https://www.example.com')),
                 asyncio.create_task(fetch_status(session, 'https://www.example.com')),
                 asyncio.create_task(fetch_status(session, 'https://www.example.com')),
    ]
  done, pending = await asyncio.wait(fetchers, return_when=asyncio.FIRST_COMPLETED)
  for done_task in done:
    result = await done_task
asyncio.run(main())

import asyncio
import logging

from aiohttp import ClientSession
from util import fetch_status

async def main() -> None:
  async with ClientSession as session:
    fetchers = [ asyncio.create_task(fetch_status(session, 'https://www.example.com')),
                 asyncio.create_task(fetch_status(session, 'https://www.example.com')),
                 asyncio.create_task(fetch_status(session, 'https://www.example.com', delay=3)),
    ]
  done, pending = await asyncio.wait(fetchers, timeout=1)
  for done_task in done:
    result = await done_task
asyncio.run(main())

import asyncio
import os
import shutil
import sys
import tty

from asyncio import StreamReader
from collections import deque
from typing import Callable, Deque, Awaitable


async def create_stdin_reader() -> StreamReader:
 stream_reader = asyncio.StreamReader()
 protocol = asyncio.StreamReaderProtocol(stream_reader)
 loop = asyncio.get_running_loop()
 await loop.connect_read_pipe(lambda: protocol, sys.stdin)
 return stream_reader

async def read_line(stdin_reader: StreamReader) -> str:
 def erase_last_char():
  move_back_one_char()
  sys.stdout.write(' ')
  move_back_one_char()
 delete_char = b'\x7f'
 input_buffer = deque()
 while ( input_char := await stdin_reader.read(1)) != b'\n':
  if input_char == delete_char:
   if len(input_buffer) != 0:
    input_buffer.pop()
    erase_last_char()
    sys.stdout.flush()
  else:
   input_buffer.append(input_char)
   sys.stdout.write(input_char.decode())
   sys.stdout.flush()
 clear_line()
 return b''.join(input_buffer).decode()

class MessageStore:
 def __init__(self, callback: Callable[[Deque], Awaitable[None]], max_size: int):
  self._deque = deque(maxlen=max_size)
  self._callback = callback

 async def append(self, item):
  self._deque.append(item)
  await self._callback(self._deque)

def save_cursor_position():
 sys.stdout.write('\0337')

def restore_cursor_position():
 sys.stdout.write('\0338')

def move_to_bottom_of_screen() -> int:
 _, total_rows = shutil.get_terminal_size()
 input_row = total_rows - 1
 sys.stdout.write(f'\033[{input_row}E')
 return total_rows

def move_to_top_of_screen():
 sys.stdout.write('\033[H')

def move_back_one_char():
 sys.stdout.write('\033[1D')

def delete_line():
 sys.stdout.write('\033[2K')

def clear_line():
 sys.stdout.write('\033[2K\033[0G')

async def sleep(delay: int, message_store: MessageStore):
 await message_store.append(f'Starting delay {delay}')
 await asyncio.sleep(delay)
 await message_store.append(f'Finished delay {delay}')


async def main():
 tty.setcbreak(sys.stdin)
 os.system('clear')
 rows = move_to_bottom_of_screen()

 async def redraw_output(items: deque):
  save_cursor_position()
  move_to_top_of_screen()
  for item in items:
   delete_line()
   print(item)
  restore_cursor_position()

 messages = MessageStore(redraw_output, rows-1)

 stdin_reader = await create_stdin_reader()

 while True:
  line = await read_line(stdin_reader)
  delay_time = int(line)
  asyncio.create_task(sleep(delay_time, messages))

asyncio.run(main())

async def echo_round() -> AsyncGenerator[int, float]:
    sent = yield 0
    while sent >= 0.0:
        rounded = await round(sent)
        sent = yield rounded

async def coroutine_method():
    return 3

async def generator_method():
    yield 3

# This is correct
r = await coroutine_method()

# This will raise an exception!
r = await generator_method()

import asyncio
import sys
from util.delay import delay
from util.create_stdin_reader import create_stdin_reader

async def main():
  stdin_reader = await create_stdin_reader()
  while True:
    delay_time = await stdin_reader.readline()
    asyncio.create_task(delay(int(delay_time)))

asyncio.run(main())

import asyncio
import logging
import os
import sys
import tty
from asyncio import StreamReader, StreamWriter
from collections import deque
from util import create_stdin_reader, cursor, message_store
from util.message_store import MessageStore

async def read_and_send(message: str, writer: StreamWriter):
  writer.write((message + '\n').encode())
  await writer.drain()

async def listen_for_messages(reader: StreamReader, message_store: MessageStore):
  while (message := await reader.readline()) != b'':
    await message_store.append(message.decode())
  await message_store.append('Server closed connection')


async def main():
  tty.setcbreak(0)
  os.system('clear')
  rows = cursor.move_to_bottom_of_screen()

  async def redraw_output(items: deque):
    cursor.save_cursor_position()
    cursor.move_to_top_of_screen()
    for item in items:
      cursor.delete_line()
      print(item)
    cursor.restore_cursor_position()

  messages = MessageStore(redraw_output, rows-1)
  stdin_reader = await create_stdin_reader.create_stdin_reader()
  sys.stdout.write('Enter username: ')
  username = await cursor.read_line(stdin_reader)
  
  reader, writer = await asyncio.open_connection('127.0.0.1', 8000)
  writer.write(f'CONNECT {username}\n'.encode())
  await writer.drain()

  message_listener = asyncio.create_task(listen_for_messages(reader, messages))
  input_listener = asyncio.create_task(read_and_send(stdinreader, writer))

  try:
    await asyncio.wait([message_listener, input_listener], return_when=asyncio.FIRST_COMPLETED)
  except Exception as e:
    logging.exception(e)
    writer.close()
    await writer.wait_closed()

asyncio.run(main())

import asyncio
import logging
from asyncio import StreamReader, StreamWriter

class ChatServer:
  def __init__(self):
    self._username_to_writer = {}
  
  async def start_chat_server(self, host: str, port: int):
    server = await asyncio.start_server(self.client_connected, host, port)

    async with server:
      await server.serve_forever()
  
  async def client_connected(self, reader: StreamReader, writer: StreamWriter):
    command = await reader.readline()
    print(f'CONNECTED {reader} {writer}')
    command, args = command.split(b' ')
    print(command)
    # print(args)
    if command == b'CONNECT':
      username = args.replace(b'\r\n', b'').decode()
      self._add_user(username, reader, writer)
      await self._on_connect(username, writer)
    else:
      logging.error('Got invalid command from client, disconnecting.')
      writer.close()
      await writer.wait_closed()

  def _add_user(self, username: str, reader: StreamReader, writer: StreamWriter):
    self._username_to_writer[username] = writer
    asyncio.create_task(self._listen_for_messages(username, reader))

  async def _remove_user(self, username: str):
    writer = self._username_to_writer[username]
    del self._username_to_writer[username]
    try:
      writer.close()
      await writer.wait_closed()
    except Exception as e:
      logging.exception('Error closing client writer, ignoring.', exc_info=e)

  async def _notify_all(self, message: str):
    inactive_users = []
    for username, writer in self._username_to_writer.items():
      try:
        writer.write(message.encode())
        await writer.drain()
      except ConnectionError as e:
        logging.exception('Could not write to client.', exc_info=e)
        inactive_users.append(username)
    [ await self._remove_user(username) for username in inactive_users ]

  async def _listen_for_messages(self, username: str, reader: StreamReader):
    try:
      while (data := await asyncio.wait_for(reader.readline(), 60)) != b'':
        await self._notify_all(f'{username}: {data.decode()}')
      await self._notify_all(f'{username} has left the chat\n')
    except Exception as e:
      logging.exception('Error reading from client.', exc_info=e)
      await self._remove_user(username)

  async def _on_connect(self, username: str, writer: StreamWriter):
    writer.write(f'Welcome! {len(self._username_to_writer)} user(s) are online!\n'.encode())
    await writer.drain()
    await self._notify_all(f'{username} connected!\n')

async def main():
  chat_server = ChatServer()
  await chat_server.start_chat_server('127.0.0.1', 8000)

asyncio.run(main())

## The code creates some errors

import asyncio
import asyncpg
import os
import shutil
import sys
import tty

from collections import deque
from asyncio import StreamReader
from asyncpg import Pool
from typing import Callable, Deque, Awaitable

async def create_stdin_reader() -> StreamReader:
 stream_reader = asyncio.StreamReader()
 protocol = asyncio.StreamReaderProtocol(stream_reader)
 loop = asyncio.get_running_loop()
 await loop.connect_read_pipe(lambda: protocol, sys.stdin)
 return stream_reader

class MessageStore:
 def __init__(self, callback: Callable[[Deque], Awaitable[None]], max_size: int):
  self._deque = deque(maxlen=max_size)
  self._callback = callback

 async def append(self, item):
  self._deque.append(item)
  await self._callback(self._deque)

def save_cursor_position():
 sys.stdout.write('\0337')

def restore_cursor_position():
 sys.stdout.write('\0338')

def move_to_bottom_of_screen() -> int:
 _, total_rows = shutil.get_terminal_size()
 input_row = total_rows - 1
 sys.stdout.write(f'\033[{input_row}E')
 return total_rows

def move_to_top_of_screen():
 sys.stdout.write('\033[H')

def move_back_one_char():
 sys.stdout.write('\033[1D')

def delete_line():
 sys.stdout.write('\033[2K')

def clear_line():
 sys.stdout.write('\033[2K\033[0G')


async def run_query(query: str, pool: Pool, message_store: MessageStore):
 async with pool.acquire() as connection:
  try:
   result = await connection.fetchrow(query)
   await message_store.append(f'Fetched {len(result)} rows from: {query}')
  except Exception as e:
   await message_store.append(f'Got exception {e}: from {query}')

async def read_line(stdin_reader: StreamReader) -> str:
 def erase_last_char():
  move_back_one_char()
  sys.stdout.write(' ')
  move_back_one_char()
 delete_char = b'\x7f'
 input_buffer = deque()
 while ( input_char := await stdin_reader.read(1)) != b'\n':
  if input_char == delete_char:
   if len(input_buffer) != 0:
    input_buffer.pop()
    erase_last_char()
    sys.stdout.flush()
  else:
   input_buffer.append(input_char)
   sys.stdout.write(input_char.decode())
   sys.stdout.flush()
 clear_line()
 return b''.join(input_buffer).decode()

async def main():
 tty.setcbreak(0)
 os.system('clear')
 rows = move_to_bottom_of_screen()

 async def redraw_output(items: deque):
  save_cursor_position()
  move_to_top_of_screen()
  for item in items:
   delete_line()
   print(item)
  restore_cursor_position()

 messages = MessageStore(redraw_output, rows - 1)
 stdin_reader = await create_stdin_reader()

 async with asyncpg.create_pool( host='127.0.0.1',
         port=5432,
         user='postgres',
         password='password',
         database='products',
         min_size=6,
         max_size=6) as pool:
  while True:
   query = await read_line(stdin_reader)
   asyncio.create_task(run_query(query, pool, messages))

asyncio.run(main())

import asyncio
import logging
import signal
import socket

from asyncio import AbstractEventLoop
from typing import List

async def echo(connection: socket, loop: AbstractEventLoop) -> None:
  try:
    while data := await loop.sock_recv(connection, 1024): # loop forever waiting for data from a client connection
      if data == b'boom\r\n':
        raise Exception("Unexpected network error") # a line from the output: Task exception was never retrieved
      # no exception is thrown to call stack and no cleanup here `await` is required
      await loop.sock_sendall(connection, data)
  except Exception as ex:
    logging.exception(ex)
  finally:
    connection.close()

echo_tasks = []

async def connection_listener(server_socket: socket, loop: AbstractEventLoop) -> None:
  while True:
    connection, address = await loop.sock_accept(server_socket)
    connection.setblocking(False)
    print(f'Got a connection from {address}')
    echo_task = asyncio.create_task(echo(connection, loop)) # whenever we get a connection, create an echo task to listen for client data
    echo_tasks.append(echo_task)

class GracefulExit(SystemExit):
  pass

def shutdown() -> None:
  raise GracefulExit()

async def close_echo_tasks(echo_tasks: List[asyncio.Task]) -> None:
  waiters = [asyncio.wait_for(task, 2) for task in echo_tasks] # `asyncio.wait_for` sets timeouts on a task
  for task in waiters:
    try:
      await task
    except asyncio.exceptions.TimeoutError:
      pass

async def main():
  server_socket = socket.socket()
  server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

  server_address = ('127.0.0.1', 8000)
  server_socket.setblocking(False)
  server_socket.bind(server_address)
  server_socket.listen()

  for signame in [ 'SIGINT', 'SIGTERM' ]:
    loop.add_signal_handler(getattr(signal, signame), shutdown) # listen to the event, it can safely interact with the event loop
  await connection_listener(server_socket, loop)

loop = asyncio.new_event_loop() # create a new event loop

try:
  loop.run_until_complete(main()) # take a coroutine and run it until it finishes
except GracefulExit:
  loop.run_until_complete(close_echo_tasks(echo_tasks))
finally:
  loop.close()

"""
# `response` is a `Response` object, with which we can get
# all the information we need from this object.
"""
import requests

# `GET` retrieves resources from a given API. It is read-only.
response = requests.get('<url>') 

# `HEAD` asks for a response identical to that of `GET`,
# but only returns the status_code or HTTP header.
response = requests.head('<url>')
response.status_code # 200
response.headers["Content-Type"] # 'application/json; charset=utf-8'

# `POST` creates a new resource.
response = requests.post('<url>', data = {'key':'value'})

# `PUT` updates an existing resource.
response = requests.put('<url>', data = {'key':'value'})

# `DELETE` deletes a resource.
response = requests.delete('<url>')

# `OPTIONS` sends a `OPTIONS` request.
response = requests.options('<url>')

# `PATCH` partially updates an existing resource.
responce = requests.patch('<url>', params={<key>: <value>}, <args>)

# Passing parameters in URLs
payload = {'key1': 'value1', 'key2': ['value2', 'value3']}
response = requests.get('httpbin.org/get?key=val', params=payload)
print(r.url) # httpbin.org/get?key1=value1&key2=value2&key2=value3

# Response content
print(response.text)

# Encoding of the response based on the HTTP headers
response.encoding # 'utf-8'
response.encoding = 'ISO-8859-1' # changes the encoding

# JSON response content
response = requests.get('<url>')
response.json() # If decoding fails, it raises an exception.

import requests
from requests.exceptions import HTTPError

for url in ['https://api.github.com', 'https://api.github.com/invalid']:
  try:
    response = requests.get(url)
    response.raise_for_status() # it will be raised in case of an error
  except HTTPError as http_err:
    print(f'HTTP error occurred: {http_err}')
  except Exception as err:
    print(f'Other error occurred: {err}')
  else:
    print('Success')

"""
A signal handler that cancels all currently running tasks.
"""

import asyncio
import signal

from asyncio import AbstractEventLoop
from typing import Set
from util.delay import delay

def cancel_tasks():
  print('Got a SIGINT')
  tasks : Set[asyncio.Task] = asyncio.all_tasks()
  print(f'Cancelling {len(tasks)} task(s).')
  [task.cancel() for task in tasks]

async def main():
  loop : AbstractEventLoop = asyncio.get_running_loop()
  loop.add_signal_handler(signal.SIGINT, cancel_tasks)
  await delay(10)

asyncio.run(main())

import selectors
import socket
from selectors import SelectorKey
from typing import List, Tuple

selector = selectors.DefaultSelector()

# create a TCP socket server
# server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # socket.AF_INET := type of address, here the hostname and port number # socket.SOCK_STREAM := TCP protocol will be used
server_socket = socket.socket()
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # socket.SO_REUSEADDR, 1 := reuse the port number after stopping and restarting the application
server_socket.setblocking(False)

server_address = ('127.0.0.1', 8000) 
server_socket.bind(server_address) # clients will use 127.0.0.1 to send data to our server
server_socket.listen() # allows clients to connect to our server socket

selector.register(server_socket, selectors.EVENT_READ)

while True:
  events: List[Tuple[SelectorKey, int]] = selector.select(timeout=1) # create a selector that will timeout after 1 sec

  if len(events) == 0:
    print('No events, waiting a bit more!')
  
  for event, _ in events:

    event_socket = event.fileobj # gets the socket for the event, which is stored in the fileobj field

    if event_socket == server_socket: # This is a connection attempt.
      connection, client_address = server_socket.accept() # wait for a connection by blocking until we get a communication # connection:= another socket that we read data from and write data to our client # client_address:= the address of the client that connected
      connection.setblocking(False)
      print(f'I got a connection from {client_address}!')
      selector.register(connection, selectors.EVENT_READ) # register the client that connected with our selector

    else:
      data = event_socket.recv(1024) # receive data from the client, and echo it back
      print(f"I got some data: {data}")
      event_socket.send(data)

from multiprocess import Process
import os

def hello_from_process():
 print(f"Hello from child process {os.getpid()}")

if __name__ == "__main__":

 # `Process(target=<function>, args=<parameters_of_the_function>`
 hello_process = Process(target=hello_from_process)
 hello_process.start()
 print(f"Hello from process {os.getpid()}")
 hello_process.join() # `join` blocks the main process until each process has finished

import os, threading

def hello_from_thread():
  print(f"Hello from thread {threading.current_thread()}!")

hello_thread = threading.Thread(target=hello_from_thread)
hello_thread.start()

print(f"Python process running with process id: {os.getpid()}")

total_threads = threading.active_count()
thread_name = threading.current_thread().name

print(f"Python is currently running {total_threads} thread(s).")
print(f"The current thread is {thread_name}.")

# join will cause the program to pause until the thread we started is completed.
hello_thread.join()

import concurrent.futures

## Executor manages all the tasks that are running, either in separate processes or threads.
with concurrent.futures.Executor() as executor:
    futures = {
        executor.submit(perform, task) for task in get_tasks_to_do()
    }

    for future in concurrent.futures.as_completed(futures):
        print(f"The outcome is {future.result()}")

import asyncio
from concurrent.futures import ProcessPoolExecutor
from functools import partial
from typing import List

def count(count_to: int) -> int:
  counter = 0
  while counter < count_to:
    counter += 1
  return counter

async def main():
  with ProcessPoolExecutor as process_pool:
    loop: AbstractEventLoop = asyncio.get_running_loop()
    numbers = [ 1, 3, 5, 22, 100000000]
    calls: List[partial[int]] = [ partial(count, num) for num in nums ]
    call_coros = []

    for call in calls:
      call_coros.append(loop.run_in_executor(process_pool, call))
    results = await asyncio.gather(*call_coros)

    for result in results:
      print(result)

if __name__ == "__main__":
  asyncio.run(main())

import asyncio
import functools
import hashlib
import os
import random
import string
from concurrent.futures.thread import ThreadPoolExecutor

def random_password(length: int) -> bytes:
  ascii_lowercase = string.ascii_lowercase.encode()
  return b''.join(bytes(random.choice(ascii_lowercase)) for _ in range(length))

passwords = [ random_password(10) for _ in range(10000) ]

def hash(password: bytes) -> str:
  salt = os.urandom(16)
  return str(hashlib.scrypt(password, salt=salt, n=2048, p=1, r=8))


async def main():
  loop = asyncio.get_running_loop()
  tasks = []
  with ThreadPoolExecutor() as pool:
    for password in passwords:
      tasks.append(loop.run_in_executor(pool, functools.partial(hash, password)))
  await asyncio.gather(*tasks)

asyncio.run(main())

import asyncio
import functools
import numpy as np
from concurrent.futures.thread import ThreadPoolExecutor

data_points = 4000000000
rows = 50
columns = int(data_points / rows)
matrix = np.arange(data_points).reshape(rows, columns)

def mean_for_row(arr, row):
  return np.mean(arr[row])


async def main():
  loop = asyncio.get_running_loop()
  with ThreadPoolExecutor() as pool:
    tasks = []
    for i in range(rows):
      mean = functools.partial(mean_for_row, matrix, i)
      tasks.append(loop.run_in_executor(pool, mean))
    results = asyncio.gather(*tasks)

asyncio.run(main())

from threading import Thread
import socket


class ClientEchoThread(Thread): # inherits from `Thread` class

 def __init__(self, client):
  super().__init__()
  self.client = client

 def run(self):
  try:
   while True:
    data = self.client.recv(2048)
    if not data: # when the connection was shut down or closed by the client
     raise BrokenPipeError('Connection closed!')
    print(f'Received {data}, sending!')
    self.client.sendall(data)
  except OSError as e:
   print(f'Thread interrupted by {e} exception, shutting down!')

 def close(self):
  if self.is_alive(): # shut down the connection if the thread is alive; the thread may not be alive if the client closed the connection
   self.client.sendall(bytes('Shutting down!', encoding='utf-8'))
   self.client.shutdown(socket.SHUT_RDWR) # shut down the client connection for reads and writes


with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:

 server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 server.bind(('127.0.0.1', 8000))
 server.listen()

 connection_threads = []
 try:
  while True: # listen for connections on our server socket
   connection, _ = server.accept() # block waiting for a client to connect
   thread = ClientEchoThread(connection)
   connection_threads.append(thread) 
   thread.start()
 except KeyboardInterrupt:
  print('Shutting down!')
  [ thread.close() for thread in connection_threads ]

import asyncpg
import asyncio

from asyncpg import Record
from typing import List

CREATE_BRAND_TABLE = \
  """
  CREATE TABLE IF NOT EXISTS brand(
    brand_id SERIAL PRIMARY KEY,
    brand_name TEXT NOT NULL
  );"""

CREATE_PRODUCT_TABLE = \
  """
  CREATE TABLE IF NOT EXISTS product(
    product_id SERIAL PRIMARY KEY,
    product_name TEXT NOT NULL,
    brand_id INT NOT NULL,
    FOREIGN KEY (brand_id) REFERENCES brand(brand_id)
  );"""

CREATE_PRODUCT_COLOR_TABLE = \
  """
  CREATE TABLE IF NOT EXISTS product_color(
    product_color_id SERIAL PRIMARY KEY,
    product_color_name TEXT NOT NULL
  );"""

CREATE_PRODUCT_SIZE_TABLE = \
  """
  CREATE TABLE IF NOT EXISTS product_size(
    product_size_id SERIAL PRIMARY KEY,
    product_size_name TEXT NOT NULL
  );"""

CREATE_SKU_TABLE = \
  """
  CREATE TABLE IF NOT EXISTS sku(
    sku_id SERIAL PRIMARY KEY,
    product_id INT NOT NULL,
    product_size_id INT NOT NULL,
    product_color_id INT NOT NULL,
    FOREIGN KEY (product_id) REFERENCES product(product_id),
    FOREIGN KEY (product_size_id) REFERENCES product_size(product_size_id),
    FOREIGN KEY (product_color_id) REFERENCES product_color(product_color_id)
  );"""

COLOR_INSERT = \
  """
  INSERT INTO product_color VALUES(1, 'Blue');
  INSERT INTO product_color VALUES(2, 'Black');
  """
SIZE_INSERT = \
  """
  INSERT INTO product_size VALUES(1, 'Small');
  INSERT INTO product_size VALUES(2, 'Medium');
  INSERT INTO product_size VALUES(3, 'Large');
  """

async def main():
  connection = await asyncpg.connect( host='127.0.0.1',
                                      port=5432,
                                      user='postgres',
                                      database='products',
                                      password='password' )

  statements = [  CREATE_BRAND_TABLE,
                  CREATE_PRODUCT_TABLE,
                  CREATE_PRODUCT_COLOR_TABLE,
                  CREATE_PRODUCT_SIZE_TABLE,
                  CREATE_SKU_TABLE,
                  SIZE_INSERT,
                  COLOR_INSERT ]

  print('Creating the product database...')
  for statement in statements:
    status = await connection.execute(statement)
    print(status)
  print('Finished creating the product database!')

  await connection.execute("INSERT INTO brand VALUES(DEFAULT, 'Levis')")
  await connection.execute("INSERT INTO brand VALUES(DEFAULT, 'Seven')")

  results : List[Record] = await connection.fetch('SELECT brand_id, brand_name FROM brand')
  for brand in results:
    print(f'id: {brand["brand_id"]}, name: {brand["brand_name"]}')

  await connection.close()

asyncio.run(main())

import asyncio
import asyncpg

from typing import Tuple, Union, List
from random import sample

def load_common_words() -> List[str]:
  with open('/Users/macos/Desktop/python/concurrency/common_words.txt') as common_words:
    return common_words.readlines()

def generate_brand_names(words: List[str]) -> List[Tuple[Union[str, ]]]:
  return [(words[index],) for index in sample(range(100), 100)]

async def insert_brands(common_words, connection) -> int:
  brands = generate_brand_names(common_words)
  insert_brands = "INSERT INTO brand VALUES(DEFAULT, $1)"
  return await connection.executemany(insert_brands, brands)

async def main() -> None:
  common_words = load_common_words()
  connection = await asyncpg.connect( host='127.0.0.1',
                                      port=5432,
                                      user='postgres',
                                      database='products',
                                      password='password' )
  await insert_brands(common_words, connection)

asyncio.run(main())

import asyncio
import asyncpg

from typing import List, Tuple
from random import sample, randint

def load_common_words() -> List[str]:
  with open('/Users/macos/Desktop/python/concurrency/common_words.txt') as file:
    return file.readlines()

def gen_products( common_words: List[str],
                  brand_id_start: int,
                  brand_id_end: int,
                  products_to_create: int ) -> List[ Tuple[str, int] ]:
  products = []
  for _ in range(products_to_create):
    description = [ common_words[index] for index in sample(range(1000), 10) ]
    brand_id = randint(brand_id_start, brand_id_end)
    products.append((" ".join(description), brand_id))
  return products

def gen_skus( product_id_start: int,
              product_id_end: int,
              skus_to_create: int ) -> List[Tuple[int, int, int]]:
  skus = []
  for _ in range(skus_to_create):
    product_id = randint(product_id_start, product_id_end)
    size_id = randint(1,3)
    color_id = randint(1,2)
    skus.append((product_id, size_id, color_id))
  return skus

async def main() -> None:
  common_words = load_common_words()
  connection = await asyncpg.connect( host='127.0.0.1',
                                      port=5432,
                                      user='postgres',
                                      database='products',
                                      password='password' )
  product_tuples = gen_products( common_words,
                                  brand_id_start=1,
                                  brand_id_end=100,
                                  products_to_create=1000 )
  await connection.executemany("INSERT INTO product VALUES(DEFAULT, $1, $2)", product_tuples)
  sku_tuples = gen_skus( product_id_start=1,
                         product_id_end=1000,
                         skus_to_create=100000 )
  await connection.executemany("INSERT INTO sku VALUES(DEFAULT, $1, $2, $3)", sku_tuples)
  await connection.close()
asyncio.run(main())

import asyncio
import asyncpg

product_query = \
"""
SELECT p.product_id, p.product_name, p.brand_id, s.sku_id, pc.product_color_name, ps.product_size_name
FROM product as p
JOIN sku as s on s.product_id = p.product_id
JOIN product_color as pc on pc.product_color_id = s.product_color_id
JOIN product_size as ps on ps.product_size_id = s.product_size_id
WHERE p.product_id = 100
"""

async def query_product(pool):
  async with pool.acquire() as connection:
    return await connection.fetchrow(product_query)

async def main() -> None:
  async with asyncpg.create_pool( host='127.0.0.1',
                                  port=5432,
                                  user='postgres',
                                  password='password',
                                  database='products',
                                  min_size=6,
                                  max_size=6 ) as pool:
  
    # execute two product queries concurrently
    await asyncio.gather(query_product(pool), query_product(pool))

asyncio.run(main())

import asyncio
import asyncpg
import logging

async def main():
  connection = await asyncpg.connect( host='127.0.0.1',
                                      database='products',
                                      username='postgres',
                                      password='password',
                                      port=5432 )
  try:
    async with connection.transaction():
      insert_brand = "INSERT INTO brand VALUES(9999, 'big_brand')"
      await connection.execute(insert_brand)
      await connection.execute(insert_brand) # an exception will be raise due to duplicate primary key
  except Exception:
    logging.exception('Error while running transaction')
  finally:
    query = "SELECT brand_name FROM brand WHERE brand_name LIKE 'big_%'"
    brands = await connection.fetch(query)
    print(brands) # [] := transaction is rolled back.

  await connection.close()

asyncio.run(main())

import asyncio
import asyncpg

from asyncpg.transaction import Transaction

async def main():
  connection = await asyncpg.connect( host='127.0.0.1',
                                      database='products',
                                      username='postgres',
                                      password='password',
                                      port=5432 )
  transaction : Transaction = connection.transaction() # creates a transaction instance
  await transaction.start() # start the transaction
  try:
    await connection.execute("INSERT INTO brand VALUES(DEFAULT, 'brand_1')")
    await connection.execute("INSERT INTO brand VALUES(DEFAULT, 'brand_2')")
  except asyncpg.PostgresError:
    print('Errors, rolling back transaction!')
    await transaction.rollback()
  else:
    print('No errors, committing transaction!')
    await transaction.commit()

  query = "SELECT brand_name FROM brand WHERE brand_name LIKE 'big_%'"
  brands = await connection.fetch(query)
  print(brands) # [] := transaction is rolled back.

  await connection.close()

asyncio.run(main())

import asyncio
import asyncpg

async def main():
  connection = await asyncpg.connect( host='127.0.0.1',
                                      database='products',
                                      username='postgres',
                                      password='password',
                                      port=5432 )
  async with connection.transaction():
    query = 'SELECT product_id, product_name FROM product'
    ## async for ##
    # async for product in connection.cursor(query):
    #   print(product)
    ###############
    cursor = await connection.cursor(query) # creates a cursor for the query
    await cursor.forward(500) # move the cursor forward 500 records
    products = await cursor.fetch(100) # get the next 100 records
    for product in products:
      print(product)

  await connection.close()

asyncio.run(main())

import asyncio
import asyncpg

async def take(generator, to_take: int):
  item_count = 0
  async for item in generator:
    if item_count > to_take - 1:
      return
    item_count += 1
    yield item

async def main():
  connection = await asyncpg.connect( host='127.0.0.1',
                                      database='products',
                                      username='postgres',
                                      password='password',
                                      port=5432 )
  async with connection.transaction():
    query = 'SELECT product_id, product_name FROM product'
    product_generator = await connection.cursor(query)
    async for product in take(product_generator, 5):
      print(product)
    print('Got the first 5 products')
  
  await connection.close()

asyncio.run(main())

import asyncio
from asyncio import AbstractEventLoop, Future, Transport
from typing import Optional

class HTTPGetClientProtocol(asyncio.Protocol): # extends asyncio.Protocol class

 def __init__(self, host: str, loop: AbstractEventLoop):
  self._host: str = host
  self._future: Future = loop.create_future()
  self._transport: Optional[Transport] = None # transport communicates with the server
  self._response_buffer: bytes = b''

 async def get_response(self):
  return await self._future

 def _get_request_bytes(self) -> bytes:
  request = f"GET / HTTP/1.1\r\n" \
        f"Connection: close\r\n" \
            f"Host: {self._host}\r\n\r\n"
  return request.encode()

 def connection_made(self, transport: Transport):
  """ `Transport` calls it when the underlying socket has successfully
  connected with the HTTP server.
  """
  print(f'Connection made to {self._host}')
  self._transport = transport
  self._transport.write(self._get_request_bytes())

 def data_received(self, data):
  """ `Transport` calls it whenever it receives data,
  passing it to us as bytes.
  """
  print('Data received!')
  self._response_buffer += data # _response_buffer can be called multiple times

 def eof_received(self):
  """Calling this method guarantees that data_received
  will never be called again.
  """
  self._future.set_result(self._response_buffer.decode())
  return False

 def connection_lost(self, exc: Optional[Exception]) -> None:
  if exc is None:
   print('Connection closed without error.')
  else:
   self._future.set_exception(exc)

async def make_request(host: str, port: int, loop: AbstractEventLoop) -> str:
 def protocol_factory():
  return HTTPGetClientProtocol(host, loop)

 _, protocol = await loop.create_connection(protocol_factory, host=host, port=port)
 return await protocol.get_response()

async def main():
 loop = asyncio.get_running_loop()
 result = await make_request('www.example.com', 80, loop)
 print(result)

asyncio.run(main())

import asyncio
import logging

from asyncio import StreamWriter, StreamReader

class ServerState:

  def __init__(self):
    self._writers = []
  
  async def add_client(self, reader: StreamReader, writer: StreamWriter):
    """send messages to all connected clients and
    remove it when a client disconnects
    """
    self._writers.append(writer)
    await self._on_connect(writer)
    asyncio.create_task(self._echo(reader, writer))

  async def _on_connect(self, writer: StreamWriter):
    """sends a message to the client informing them how many other
    users are connected and notify any other connected clients that
    the new user has connected
    """
    writer.write(f'Welcome! {len(self._writers)} user(s) are online!\n'.encode())
    await writer.drain()
    await self._notify_all('New user connected!\n')
  
  async def _echo(self, reader: StreamReader, writer: StreamWriter):
    """When a user disconnects, we notify any other connected clients
    that someone disconnected
    """
    try:
      while (data := await reader.readline()) != b'':
        writer.write(data)
        await writer.drain()
      self._writers.remove(writer)
      await self._notify_all((f'Client disconnected. {len(self._writers)} user(s) are online!\n'))
    except Exception as e:
      logging.exception('Error reading from client.', exc_info=e)
      self._writers.remove(writer)

  async def _notify_all(self, message: str):
    for writer in self._writers:
      try:
        writer.write(message.encode())
        await writer.drain()
      except ConnectionError as e:
        logging.exception('Could not write to client.', exc_info=e)
        self._writers.remove(writer)

async def main():
  server_state = ServerState()
  
  async def client_connected(reader: StreamReader, writer: StreamWriter) -> None:
    await server_state.add_client(reader, writer)

  # start server and start serving forever
  server = await asyncio.start_server(client_connected, '127.0.0.1', 8000)

  async with server:
    await server.serve_forever()

asyncio.run(main())

import asyncio
from asyncio import StreamReader
from typing import AsyncGenerator

async def read_until_empty(stream_reader: StreamReader) -> AsyncGenerator[str, None]:
 while response := await stream_reader.readline():
  yield response.decode()

async def main():
 host: str = 'www.example.com'
 request: str = f"GET / HTTP/1.1\r\n" \
       f"Connection: close\r\n" \
       f"Host: {host}\r\n\r\n"

 stream_reader, stream_writer = await asyncio.open_connection(host, 80)

 try:
  stream_writer.write(request.encode())
  await stream_writer.drain()
  responses = [ response async for response in read_until_empty(stream_reader) ]
  print(''.join(responses))

 finally:
  stream_writer.close()
  await stream_writer.wait_closed()

asyncio.run(main())

import asyncio
import concurrent.futures
import functools
import time
from typing import Dict, List


def partition(data: List, chunk_size: int) -> List:
 for i in range(0, len(data), chunk_size):
  yield data[i:i+chunk_size]

def map_frequencies(chunk: List[str]) -> Dict[str, int]:
 counter = {}
 for line in chunk:
  word, _, count, _ = line.split('\t')
  if counter.get(word):
   counter[word] += int(count)
  else:
   counter[word] = int(count)
 return counter

def merge_dictionaries(first: Dict[str, int], second: Dict[str, int]) -> Dict[str, int]:
 merged = first
 for key in second:
  if key in merged:
   merged[key] += second[key]
  else:
   merged[key] = second[key]
 return merged

async def reduce(loop, pool, counters, chunk_size) -> Dict[str, int]:
 chunks: List[List[Dict]] = list(partition(counters, chunk_size))
 reducers = []
 while len(chunks[0]) > 1:
  for chunk in chunks:
   reducer = functools.partial(functools.reduce, merge_dictionaries, chunk)
   reducers.append(loop.run_in_executor(pool, reducer))
  reducer_chunks = await asyncio.gather(*reducers)
  chunks = list(partition(reducer_chunks, chunk_size))
  reducers.clear()
 return chunks[0][0]


async def main(partition_size: int):
 with open('googlebooks-fre-all-1gram-20120701-a', encoding='utf-8') as file:
  contents = file.readlines()
  loop = asyncio.get_running_loop()
  tasks = []
  with concurrent.futures.ProcessPoolExecutor() as pool:
   for chunk in partition(contents, partition_size):
    tasks.append(loop.run_in_executor(pool, functools.partial(map_frequencies, chunk)))
   intermediate_results = await asyncio.gather(*tasks)
   final_result = await reduce(loop, pool, intermediate_results, 500)
   print(f"Aardvark has appeared {final_result['Aardvark']} times.")


if __name__ == "__main__":
 asyncio.run(main(partition_size=60000))

import asyncio
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Value

shared_counter: Value

def init(counter: Value):
  global shared_counter
  shared_counter = counter

def increment():
  with shared_counter.get_lock():
    shared_counter.value += 1

async def main():
  counter = Value('d', 0)
  with ProcessPoolExecutor(initializer=init, initargs=(counter,)) as pool:
    await asyncio.get_running_loop().run_in_executor(pool, increment)
    print(counter.value)

if __name__ == "__main__":
  asyncio.run(main())

import asyncio
import requests
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from threading import Lock

counter_lock = Lock()
counter: int = 0

def get_status_code(url: str) -> int:
  global counter
  response = requests.get(url)
  with counter_lock:
    counter += 1
  return response.status_code

async def reporter(request_count: int):
  while counter < request_count:
    print(f'Finished {counter}/{request_count} requests')
    await asyncio.sleep(.5)


async def main():
  loop = asyncio.get_running_loop()
  with ThreadPoolExecutor() as pool:
    request_count = 200
    urls = ['https://www.example.com' for _ in range(request_count)]
    reporter_task = asyncio.create_task(reporter(request_count))
    tasks = [loop.run_in_executor(pool, partial(get_status_code, url)) for url in urls]
    results = await asyncio.gather(*tasks)
    await reporter_task
    print(results)

asyncio.run(main())

import asyncio
import requests

def get_status_code(url: str) -> int:
 response = requests.get(url)
 return response.status_code


async def main():
 urls = ['https://www.example.com' for _ in range(1000)]
 tasks = [asyncio.to_thread(get_status_code, url) for url in urls]
 results = await asyncio.gather(*tasks)
 print(results)

asyncio.run(main())

import asyncio
import sys
from asyncio import Lock
from util.delay import delay

async def a(lock: Lock):
  print('Coroutine a waiting to acquire the lock')
  async with lock:
    print('Coroutine a is in the critical section')
    await delay(5)
  print('Coroutine a released the lock')

async def b(lock: Lock):
  print('Coroutine b waiting to acquire the lock')
  async with lock:
    print('Coroutine b is in the critical section')
    await delay(2)
  print('Coroutine b released the lock')


async def main():
  sys.path.append('../util')
  lock = Lock()
  await asyncio.gather(a(lock), b(lock))

asyncio.run(main())

import functools
import time
from typing import Callable, Any

def async_timed():

  def wrapper(func: Callable) -> Callable:

    @functools.wraps(func)
    async def wrapped(*args, **kwargs) -> Any:
      print(f'starting {func} with args {args} {kwargs}')
      start = time.time()

      try:
        return await func(*args, **kwargs)
      finally:
        end = time.time()
        total = end - start
        print(f'finished {func} in {total:.4f} second(s)')

    return wrapped

  return wrapper

import asyncio
import sys
from asyncio import StreamReader

async def create_stdin_reader() -> StreamReader:
  stream_reader = asyncio.StreamReader()
  protocol = asyncio.StreamReaderProtocol(stream_reader)
  loop = asyncio.get_running_loop()
  await loop.connect_read_pipe(lambda: protocol, sys.stdin)
  return stream_reader

import shutil
import sys
from asyncio import StreamReader
from collections import deque

def save_cursor_position():
 sys.stdout.write('\0337')

def restore_cursor_position():
 sys.stdout.write('\0338')

def move_to_bottom_of_screen() -> int:
 _, total_rows = shutil.get_terminal_size()
 input_row = total_rows - 1
 sys.stdout.write(f'\033[{input_row}E')
 return total_rows

def move_to_top_of_screen():
 sys.stdout.write('\033[H')

def move_back_one_char():
 sys.stdout.write('\033[1D')

def delete_line():
 sys.stdout.write('\033[2K')

def clear_line():
 sys.stdout.write('\033[2K\033[0G')

async def read_line(stdin_reader: StreamReader) -> str:

 def erase_last_char():
  move_back_one_char()
  sys.stdout.write(' ')
  move_back_one_char()

 delete_char = b'\x7f'
 input_buffer = deque()
 while ( input_char := await stdin_reader.read(1)) != b'\n':
  if input_char == delete_char:
   if len(input_buffer) != 0:
    input_buffer.pop()
    erase_last_char()
    sys.stdout.flush()
  else:
   input_buffer.append(input_char)
   sys.stdout.write(input_char.decode())
   sys.stdout.flush()
 clear_line()
 return b''.join(input_buffer).decode()

"""
It will help us see what other code, if any,
is running concurrently while our coroutines
are paused.
"""
import asyncio

async def delay(delay_seconds: int) -> int:
 print(f'sleeping for {delay_seconds} second(s)')
 await asyncio.sleep(delay_seconds)
 print(f'finished sleeping for {delay_seconds} second(s)')
 return delay_seconds

import asyncio
from aiohttp import ClientSession, ClientTimeout

async def fetch_status(session: ClientSession, url: str) -> int:

  ten_millis = ClientTimeout(total=.01)

  async with session.get(url, timeout=ten_millis) as result: # we can process the result within this block
    return result.status

from collections import deque
from typing import Awaitable, Callable, Deque

class MessageStore:
  def __init__(self, callback: Callable[[Deque], Awaitable[None]], max_size: int):
    self._deque = deque(maxlen=max_size)
    self._callback = callback

  async def append(self, item):
    self._deque.append(item)
    await self._callback(self._deque)