Skip to content

Instantly share code, notes, and snippets.

@tuxdna

tuxdna/README.md

Last active November 21, 2024 06:28
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save tuxdna/2abe7f922a5704cffce6740582828eb0 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save tuxdna/2abe7f922a5704cffce6740582828eb0 to your computer and use it in GitHub Desktop.
Download ZIP
Gevent examples
Raw
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))
Raw
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()
Raw
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()
Raw
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)
Raw
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)
Raw
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}")
Raw
pseudo_thread.py
import gevent
g = gevent.spawn(1 / 0)
g.join()
gevent.sleep(3)
Raw
requirements.txt
requests
gevent
web.py
Raw
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}")
Raw
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")
Raw
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()
Raw
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()
Raw
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()
Raw
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)
Raw
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment