Gevent examples

README.md

Examples from this Youtube Playlist

https://www.youtube.com/watch?v=6rkyHJouYAQ&list=PLIO3UV9ODwNCJOH_EzE_T_fK4-g-G-xKf&index=2

context_example.py

from gevent import Greenlet

def inside():
    print("We are inside the context mamanger!")
    return -10

with Greenlet.spawn(inside) as g:
    print(f"Entered the context manager!")

print(g.get(block=False))

file_socket_client.py

import socket

s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.connect('./file_socket_server.py.sock')
s.send('GET / HTTP/1.0\r\n\r\n'.encode())
data = s.recv(1024)
print(f"We received {len(data)} bytes")
print(data.decode())
s.close()

file_socket_server.py

import os
from gevent.pywsgi import WSGIServer

from gevent import socket

def application(env, start_response):
    assert env
    start_response('200 OK', [])
    return[]

listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
socketname = "./" + os.path.basename(__file__) + ".sock"

if os.path.exists(socketname):
    os.remove(socketname)

listener.bind(socketname)
listener.listen(1)

WSGIServer(listener, application).serve_forever()

hello_gevent.py

from gevent import monkey
import gevent

monkey.patch_all()
import requests

urls = [
    'https://www.google.com',
    'https://www.apple.com',
    'https://www.python.org',
]


def response(url):
    print(f"GETing url: {url}")
    print(f"Response: {requests.get(url=url)}")


jobs = [gevent.spawn(response, url) for url in urls]

gevent.wait(jobs)

hello_subprocess_gevent.py

from gevent import monkey
import gevent

monkey.patch_all()
import requests
import subprocess

commands = [
    ['uname', '-a'],
    ['ip', 'addr'],
]


def async_command(cmd):
    print(f"Running cmd: {cmd}")
    stdout, stderr = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
    # print(f"{stdout.decode()=}\n{stderr.decode()=}")
    print(stdout.decode())
    print(stderr.decode())


jobs = [gevent.spawn(async_command, cmd) for cmd in commands]

gevent.wait(jobs)

join_kill_get.py

from gevent import Greenlet
from gevent import sleep

g = Greenlet(sleep, 5)
print(f"Starting sleep!")

g.start()
print(f"Killing immediately")

g.kill(Exception("You shall not live!"))

print(f"is dead: {g.dead}")

pseudo_thread.py

import gevent

g = gevent.spawn(1 / 0)
g.join()
gevent.sleep(3)

requirements.txt

requests
gevent
web.py

subclassing_greenlet.py

import gevent
from gevent import Greenlet
from gevent import sleep


class MyGreenlet(Greenlet):
    def __init__(self, seconds):
        Greenlet.__init__(self)
        self.seconds = seconds

    def _run(self):
        sleep(self.seconds)

    def __str__(self):
        return f"MyGreenlet({self.seconds}"


g = MyGreenlet(5)
print(f"Starting sleep!")

g.start()
print(f"Killing immediately")

g.kill(Exception("You shall not live!"))

print(f"is dead: {g.dead}")

thread_pool_examle.py

from gevent.threadpool import  ThreadPool

import gevent
import time

pool = ThreadPool(5)
start = time.time()
for _ in range(10):
    pool.spawn(time.sleep, 1)

gevent.wait()

delay = time.time() - start
print(f"Running time was: {delay} seconds")

udp_client.py

from gevent import socket
import sys

address = ("127.0.0.1", 9000)

message = " ".join(sys.argv)
sock = socket.socket(type=socket.SOCK_DGRAM)

sock.connect(address)

print(f"Sending message...")
sock.send(message.encode())

data, address = sock.recvfrom(8192)
print(f"Got: {address}, {(data,)}")
sock.close()

udp_server.py

from gevent.server import DatagramServer


class EchoServer(DatagramServer):
    def handle(self, data, address):
        print(f"Address: {address}, data: {data}")
        self.socket.sendto((f"Received: {len(data)} bytes!").encode(),
                           address)


print(f"Receiving datagrams on port 9000")
EchoServer(":9000").serve_forever()

web_app.py

from gevent import monkey

monkey.patch_all()

from gevent.pywsgi import WSGIServer
import random
import time
import web


class Index:
    def GET(self):
        return "<h1>Default context route</h1>"


class LongPolling:
    def GET(self):
        wait_secs = random.randint(1, 5)
        time.sleep(wait_secs)
        return f"<h1>Long Polling response after: {wait_secs}</h1>"


class VeryLongPolling:
    def GET(self):
        wait_secs = random.randint(10, 20)
        time.sleep(wait_secs)
        return f"<h1>Very Long Polling response after: {wait_secs}</h1>"


urls = (
    "/", Index,
    "/long", LongPolling,
    "/verylong", VeryLongPolling
)


HOST = '127.0.0.1'
PORT = 8091

application = web.application(urls).wsgifunc()
print(f"Serving on host: {HOST}, port: {PORT}")
WSGIServer((HOST, PORT), application).serve_forever()

wsgi_client.py

from gevent import monkey
import gevent

monkey.patch_all()
import requests

urls = [
    'http://127.0.0.1:8091/',
    'http://127.0.0.1:8091/users',
    'http://127.0.0.1:8091/no_url',
]


def response(url):
    print(f"GETing url: {url}")
    print(f"Response: {requests.get(url=url)}")


jobs = [gevent.spawn(response, url) for url in urls]

gevent.wait(jobs)

wsgi_server.py

from gevent.pywsgi import WSGIServer


def application(env, start_response):
    http_headers = [('Content-Type', 'text/html'), ]
    if env['PATH_INFO'] == "/":
        start_response('200 OK', http_headers)
        return [b'<b>Hello World</b>']

    elif env['PATH_INFO'] == "/users":
        start_response('200 OK', http_headers)
        return [b'<b>List of users</b>']

    start_response('404 Not Found', http_headers)
    return [b'<h1>Not Found</h1>']


HOST = '127.0.0.1'
PORT = 8091
print(f"Serving on port {PORT}")
WSGIServer((HOST, PORT), application).serve_forever()