Skip to content

Instantly share code, notes, and snippets.

View twillis's full-sized avatar

Tom Willis twillis

  • Cincinnati Ohio
View GitHub Profile
@twillis
twillis / gist:4406666
Created December 29, 2012 12:43
how to get a cursor beyond the first 1250 so that paging patterns continue to work
def cursor_from_offset(q, offset, cursor=None):
"""
the maximum cursor position you can get in one shot is
1250 (page_size=250, offset=1000), so to get a cursor
beyond that requires a bit of work.
will return None if not results and more
"""
MAX_OFFSET = 1000
if offset <= MAX_OFFSET:
@twillis
twillis / gist:5330558
Created April 7, 2013 13:46
pyzmq pub/sub example
import zmq
import random
from multiprocessing import Process
import logging
logging.basicConfig(level=logging.INFO)
class ServerProcess(Process):
def run(self):
import time
@twillis
twillis / gist:5420533
Created April 19, 2013 13:58
I needed a way to register classes used in the application and wondered how easy it would be to hook into pyramids Configurator.scan method. Turns out it's not that hard.
"""play with venusian until you have a decorator that registers
classes with request.registry["service"] """
import unittest
import venusian
from pyramid.config import Configurator
class service(object):
def __init__(self, name):
self.name = name
@twillis
twillis / gist:5525713
Last active December 17, 2015 01:08
I use this snippet to call methods on the instance for before/after insert/update/delete hooks and tablename defaults to __class__.__name__
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base, declared_attr
class HookEventHandler(object):
"""
registers itself with the mapper for whatever event name passed in
looks for corresponding event_name attribute on target and calls
it if it exists
"""
@twillis
twillis / gist:5569288
Last active December 17, 2015 06:59
service registration just works. yay.
"""
contains decorators for pyramid to discover and add various things to the registry, primarily the service decorator
"""
import venusian
from repoze.lru import lru_cache
from zope.interface import implements, Interface, Attribute, classImplements
import pyramid
from pyramid import testing
from pyramid import view
import unittest
In [1]: import logging
In [2]: from webobtoolkit import client
In [3]: logging.basicConfig(level=logging.INFO)
In [4]: URL = "http://safeway.inserts2online.com/I2O_MainFrame.jsp"; qs="pageNumber=1&drpStoreID=1954&adId=49062&adPath=SafewaySafeway05292013NorcalWeeklyAd"
In [5]: c = client.Client(client.client_pipeline(logging=True, log_level="INFO"))
;; example preferences file that I use when editing python projects
;; I typically invoke emacs like so...
;; $ cd ./to/the/project
;; $ emacs -l preferences.el ./ &
(setq tw-test-project-dir (file-truename "")) ;; absolute path to this directory
(setq bin-dir (concat tw-test-project-dir "/bin")) ;; absolute path to the bin directory
(setq tw-test-flags "--failfast") ;; test flags to pass to the test runner
(setq tw-test-runner (concat bin-dir "/" "manage_dev.py test ysn")) ;; the test runner(command to run unit tests)
(setq tw-project-dir (concat tw-test-project-dir "/src/ysn" )) ;; absolute path to the project src
import sys
import random
from timeit import timeit
try:
LISTSIZE = int(sys.argv[1])
except IndexError:
LISTSIZE = 100
SETUP = "import random;LISTSIZE=%s;expected_range = range(LISTSIZE);mydata = expected_range[:];mydata.pop(random.choice(xrange(1, len(mydata))))" % LISTSIZE
@twillis
twillis / gist:8283111
Last active January 2, 2016 09:29
example of using concurrent.futures map and as completed to submit a bunch of "jobs" and deal with the results as they complete.
from concurrent.futures import as_completed, ThreadPoolExecutor
import random
import time
def do_some_work(work_input):
result = random.choice(range(1,2))
print("sleeping for %s (%s)" % (result, work_input))
time.sleep(result)
return (result, work_input)
@twillis
twillis / gist:9712569
Created March 22, 2014 19:06
with this function, I commit the id3tags of my music collection to elasticsearch, now for the web part
(defn do-scan [initial-dir]
(dorun
(map prn (map :canonical-path
(filter identity
(map upsert-file-doc
(map file-data
(supported-files-seq initial-dir))))))))