Skip to content

Instantly share code, notes, and snippets.


Mike DePalatis mivade

View GitHub Profile
mivade /
Created Oct 26, 2018
Using watchdog with asyncio
import asyncio
from pathlib import Path
from typing import Optional
from import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer
class _EventHandler(FileSystemEventHandler):
def __init__(self, queue: asyncio.Queue, loop: asyncio.BaseEventLoop,
mivade /
Last active May 17, 2021
Using a decorator to simplify subcommand creation with argparse
from argparse import ArgumentParser
cli = ArgumentParser()
subparsers = cli.add_subparsers(dest="subcommand")
def argument(*name_or_flags, **kwargs):
"""Convenience function to properly format arguments to pass to the
subcommand decorator.
mivade /
Last active Apr 25, 2021
ZeroMQ Curve authentication demo
"""Simple demonstration of using ZMQ's Curve authentication.
This demo is adapted from the examples given in the `PyZMQ repository`__. Key
differences include:
* Using ``setsockopt`` to set Curve parameters instead of setting attributes
directly (help out your IDE!)
* Integration with ``asyncio``
mivade / benchmark.ipynb
Last active Feb 25, 2021
Sharing memory between processes with various methods
View benchmark.ipynb
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
mivade /
Last active Feb 25, 2021
Interprocess communication speed comparisons
from multiprocessing import Event, Process, Queue
import time
import zmq
class BaseActor(Process):
ready = Event()
def handle(self, msg):
print("dt =", time.time() - msg['timestamp'])
mivade /
Last active Feb 9, 2021
Tornado server-sent events
"""Demonstration of server-sent events with Tornado. To see the
stream, you can either point your browser to ``http://localhost:8080``
or use ``curl`` like so::
$ curl http://localhost:8080/events
import signal
from tornado import web, gen
mivade /
Last active Mar 17, 2020
Background tasks with tornado and concurrent.futures
"""A simple demonstration of running background tasks with Tornado.
Here I am using a basic TCP server which handles streams and keeps
them open while asynchronously performing a fake task in the
background. In order to test it, simply telnet to localhost port 8080
and start typing things to see that the server receives the messages.
The advantage to running on an executor instead of conventional
threads is that we can more easily shut it down by stopping the
tornado IO loop.
mivade / #tornado-websocket-pubsub.rst
Last active Jul 5, 2019
Using Tornado websockets for the publish/subscribe pattern
mivade /
Last active Jun 30, 2019
Websockets with Flask via aiohttp
"""Simple demo of using Flask with aiohttp via aiohttp-wsgi's
import asyncio
from aiohttp import web
from aiohttp_wsgi import WSGIHandler
from flask import Flask, render_template
class Manager(object):
classes = set()
class Meta(type):
def __new__(cls, name, bases, d):
print("name:", name, "\nbases:", bases, "\ndict: ", d)
if name is not "Base":
return type.__new__(cls, name, bases, d)