Skip to content

Instantly share code, notes, and snippets.

Avatar
👋

Mike DePalatis mivade

👋
View GitHub Profile
@mivade
mivade / aiowatch.py
Created Oct 26, 2018
Using watchdog with asyncio
View aiowatch.py
import asyncio
from pathlib import Path
from typing import Optional
from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer
class _EventHandler(FileSystemEventHandler):
def __init__(self, queue: asyncio.Queue, loop: asyncio.BaseEventLoop,
@mivade
mivade / cli.py
Last active May 17, 2021
Using a decorator to simplify subcommand creation with argparse
View cli.py
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
mivade / zmq_auth.py
Last active Apr 25, 2021
ZeroMQ Curve authentication demo
View zmq_auth.py
"""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``
__ https://github.com/zeromq/pyzmq/tree/master/examples
@mivade
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
mivade / ipc_compare.py
Last active Feb 25, 2021
Interprocess communication speed comparisons
View ipc_compare.py
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
mivade / tornadosse.py
Last active Feb 9, 2021
Tornado server-sent events
View tornadosse.py
"""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
mivade / tornadobg.py
Last active Mar 17, 2020
Background tasks with tornado and concurrent.futures
View tornadobg.py
"""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
mivade / #tornado-websocket-pubsub.rst
Last active Jul 5, 2019
Using Tornado websockets for the publish/subscribe pattern
@mivade
mivade / app.py
Last active Jun 30, 2019
Websockets with Flask via aiohttp
View app.py
"""Simple demo of using Flask with aiohttp via aiohttp-wsgi's
WSGIHandler.
"""
import asyncio
from aiohttp import web
from aiohttp_wsgi import WSGIHandler
from flask import Flask, render_template
View metaclass.py
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":
Manager.classes.add(d['data'])
return type.__new__(cls, name, bases, d)