PyDelhi 2017 content

Author: pythonhacker

pydelhi2017concurrency/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2017 Anand B Pillai
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

pydelhi2017concurrency/README.md

@@ -0,0 +1,2 @@
+# pydelhi2017concurrency
+Concurency workshop in PyDelhi 2017

pydelhi2017concurrency/async/async_fetch.py

@@ -0,0 +1,34 @@
+""" Fetch a URL using async IO """
+import asyncio
+import aiohttp
+import io
+
+async def fetch_page(url):
+    """ Asynchronous URL fetcher """
+    
+    future = aiohttp.request('GET', url)
+    # Wait for the future
+    response = await future
+    
+    return response
+    
+loop = asyncio.get_event_loop()
+urls = ('http://www.google.com',
+        'http://www.yahoo.com',
+        'http://www.facebook.com',
+        'http://www.reddit.com',
+        'http://www.twitter.com')
+
+tasks = map(fetch_page, urls)
+# Wait for tasks
+done, pending = loop.run_until_complete(asyncio.wait(tasks, timeout=120))
+loop.close()
+
+for future in done:
+    response = future.result()
+    print(response)
+    response.close()
+    
+
+
+    

pydelhi2017concurrency/async/async_fetch_url.py

@@ -0,0 +1,43 @@
+""" Fetch a URL and process its response using async io """
+
+import asyncio
+import aiohttp
+import io
+
+async def fetch_page(url):
+    """ Asynchronous URL fetcher """
+    
+    future = aiohttp.request('GET', url)
+    # Wait for the future
+    response = await future
+    
+    return response
+
+async def parse_response(done):
+    """ Parse responses of fetch """
+    
+    for future in done:
+        response = future.result()
+        data = await response.read()
+        print('Response for URL',response.url,'=>', response.status, len(data))
+        response.close()
+        
+loop = asyncio.get_event_loop()
+urls = ('http://www.google.com',
+        'http://www.yahoo.com',
+        'http://www.facebook.com',
+        'http://www.reddit.com',
+        'http://www.twitter.com')
+
+
+tasks = map(fetch_page, urls)
+# Wait for futures
+done, pending = loop.run_until_complete(asyncio.wait(tasks, timeout=120))
+# Wait for response processing
+loop.run_until_complete(parse_response(done))
+loop.close()
+
+    
+
+
+    

pydelhi2017concurrency/async/asyncio_tasks.py

@@ -0,0 +1,59 @@
+""" Co-operative multitasking example ported to asyncio """
+
+import asyncio
+import time
+
+def number_generator(m, n):
+    """ A number generator co-routine in range(m...n+1) """
+    yield from range(m, n+1)
+    
+# @asyncio.coroutine
+# async
+async def prime_filter(m, n):
+    """ Prime number co-routine """
+    
+    primes = []
+    for i in number_generator(m, n):
+        if i % 2 == 0: continue
+        flag = True
+
+        for j in range(3, int(i**0.5+1), 2):
+            if i % j == 0:
+                flag = False
+                break
+
+        if flag:
+            print('Prime=>',i)
+            primes.append(i)
+
+        # Uncomment this out and comment next line and see what happens !!!
+        await asyncio.sleep(1.0)
+
+        # time.sleep(1)         
+        
+    return tuple(primes)
+
+async def square_mapper(m, n):
+    """ Square mapper co-routine """
+    
+    squares = []
+    for i in number_generator(m, n):
+        print('Square=>',i*i)       
+        squares.append(i*i)
+        # Uncomment this out and comment next line and see what happens !!!
+        await asyncio.sleep(1.0)
+        # time.sleep(1)
+        
+    return squares
+
+def print_result(future):
+    print('Result=>',future.result())
+        
+loop = asyncio.get_event_loop()
+future = asyncio.gather(prime_filter(10, 50), square_mapper(10, 50))
+future.add_done_callback(print_result)
+loop.run_until_complete(future)
+
+loop.close()
+
+

pydelhi2017concurrency/concurrent.f/concurrent_crawler.py

@@ -0,0 +1,44 @@
+""" Simple web crawler using concurrent futures """
+
+from concurrent.futures import ProcessPoolExecutor
+from queue import Queue
+import requests
+import sys
+import urllib
+from bs4 import BeautifulSoup as bs
+
+def fetch_url(url):
+    response = requests.get(url)
+    return (response.url, response.status_code, response.content)
+    
+
+if __name__ == "__main__":
+    base_url = sys.argv[1]
+    
+    with ProcessPoolExecutor(max_workers=5) as executor:
+        urls = [base_url]
+        url_dict = {}
+        url_dict[base_url] = 1
+        
+        while len(urls):
+            results = executor.map(fetch_url, urls, timeout=120)
+            urls = []
+            
+            for result in results:
+                url, status, data = result
+                if status == 200:
+                    # Save and parse the text
+                    print('Fetched',url)
+                    soup = bs(data, "lxml")
+                    # Fetch all child links
+                    child_links = filter(None, [item.get('href', '') for item in soup.findAll('a')])
+                    # Make full links
+                    child_urls = [urllib.parse.urljoin(base_url, curl) for curl in child_links if curl[0] != '#']
+                    # Push this to queue
+                    for url in child_urls:
+                        print('Pushing',url)
+                        if not url in url_dict:
+                            url_dict[url] = 1
+                            urls.append(url)
+                    
+                

pydelhi2017concurrency/concurrent.f/concurrent_http.py

@@ -0,0 +1,21 @@
+from concurrent.futures import ProcessPoolExecutor
+import requests
+
+urls = ('http://www.google.com',
+        'http://www.yahoo.com',
+        'http://www.facebook.com',
+        'http://www.reddit.com',
+        'http://www.twitter.com')
+
+def fetch_url(url):
+    response = requests.get(url)
+    return (response.url, response.status_code, response.text)
+    
+        
+with ProcessPoolExecutor(max_workers=5) as executor:
+    results = executor.map(fetch_url, urls, timeout=120)
+    print(results)
+    for result in results:
+        url, status, data = result
+        print('Response for URL',url,'=>', status, len(data))       
+

pydelhi2017concurrency/concurrent.f/concurrent_primes.py

@@ -0,0 +1,25 @@
+from concurrent.futures import ProcessPoolExecutor, as_completed
+
+def is_prime(n):
+    """ Check for input number primality """
+    
+    for i in range(3, int(n**0.5+1), 2):
+        if n % i == 0:
+            print(n,'is not prime')
+            return False
+
+    print(n,'is prime')     
+    return True
+
+if __name__ == "__main__":
+    numbers = [1297337, 1116281, 104395303, 472882027, 533000389, 817504243, 982451653, 112272535095293, 115280095190773, 1099726899285419]*100
+
+    with ProcessPoolExecutor(max_workers=4) as executor:
+        future_map = {executor.submit(is_prime, number): number for number in numbers}
+        for future in as_completed(future_map):
+            number = future_map[future]
+            status = future.result()
+            if status:
+                print('Number',number,'is prime')
+            else:
+                print('Number',number,'aint prime')        

pydelhi2017concurrency/concurrent.f/concurrent_thumbnail.py

@@ -0,0 +1,60 @@
+"""
+
+Convert images in folder to thumbnails using concurrent futures
+
+"""
+
+import os
+import sys
+from PIL import Image
+from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
+import mimetypes
+
+def thumbnail_image(filename, size=(64,64), format='.png'):
+    """ Convert image thumbnails, given a filename """
+
+    try:
+        im=Image.open(filename)         
+        im.thumbnail(size, Image.ANTIALIAS)
+        
+        basename = os.path.basename(filename)
+        thumb_filename = os.path.join('thumbs',
+                                      basename.rsplit('.')[0] + '_thumb.png')
+        im.save(thumb_filename)
+        print('Saved',thumb_filename)
+        return True
+        
+    except Exception as e:
+        print('Error converting file',filename)
+        print(e)
+        return False
+
+def directory_walker(start_dir):
+    """ Walk a directory and generate list of valid images """
+    
+    for root,dirs,files in os.walk(os.path.expanduser(start_dir)):
+        for f in files:
+            filename = os.path.join(root,f)
+            # Only process if its a type of image
+            file_type = mimetypes.guess_type(filename.lower())[0]
+            if file_type != None and file_type.startswith('image/'):
+                yield filename
+            
+if __name__ == '__main__':
+    root_dir = os.path.expanduser('~/Pictures/')
+    if '--process' in sys.argv:
+        executor = ProcessPoolExecutor(max_workers=10)
+    else:
+        executor = ThreadPoolExecutor(max_workers=10)
+        
+    with executor:
+        future_map = {executor.submit(thumbnail_image, filename): filename for filename in directory_walker(root_dir)}
+        for future in as_completed(future_map):
+            num = future_map[future]
+            status = future.result()
+            if status:
+                print('Thumbnail of',future_map[future],'saved')
+
+
+        
+