Skip to content

Instantly share code, notes, and snippets.

View agronholm's full-sized avatar

Alex Grönholm agronholm

  • NextDay Solutions Oy
  • Nurmijärvi, Finland
View GitHub Profile
@agronholm
agronholm / reactor.py
Created January 15, 2011 04:50
reactor.py
from Queue import Queue, Empty
from socket import (socket, error, AF_INET, SOCK_STREAM, SOCK_DGRAM,
SOL_SOCKET, SOL_TCP, SO_ERROR, TCP_NODELAY)
from errno import EINPROGRESS
from select import select
from threading import current_thread
from concurrent.futures import Future
from functools import partial
import sys
import os
@agronholm
agronholm / gist:912493
Created April 10, 2011 16:31
My PyDev issues
Pydev does not see namespace packages properly (ie. "paste" of Paste Deploy, Pastescript and Paste) and considers them unresolved imports.
Code formatting messes up signed numbers ("-1" turns into "- 1").
Autocompletion for many standard library functions does not work. Try autocompleting partial (from functools) or asctime (from time).
Isn't this the entire point of specifying forced built-ins in the interpreter configuration?
There seems to be no way to get rid of error markers coming from imports that work but cannot be seen through static analysis of the code
(ie. from nose.tools import assert_raises), at least not without entirely disabling import errors.
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import *
Base = declarative_base()
class Foo(Base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
def _socket(self):
"""Create a listening socket.
This handles IPv6 and allows socket re-use by spawned processes.
"""
host, port = self.address
families = set()
for family, kind, protocol, cname, sa in socket.getaddrinfo(host or None, port, flags=socket.AI_PASSIVE):
families.add(family)
#!/usr/bin/env python
import csv
reader = csv.reader(open("xalph.txt", "rb"))
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, Session
# Set up the database.
@agronholm
agronholm / gist:6924480
Created October 10, 2013 19:53
Temporary table hack for SQLAlchemy unit tests
def create_temporary_tables(conn, cursor, statement, parameters, context, executemany):
if statement.startswith('CREATE TABLE '):
statement = 'CREATE TEMPORARY TABLE ' + statement[13:]
return statement, parameters
def setup_module():
global connection, engine
# Connect to the database and create the schema within a transaction
@agronholm
agronholm / gist:7249102
Last active October 27, 2017 14:41
Alternative Flask SQLAlchemy extension
from sqlalchemy.engine import create_engine
from sqlalchemy.orm.scoping import scoped_session
from sqlalchemy.orm.session import sessionmaker
class SQLAlchemy:
def __init__(self, app=None):
if app is not None:
self.init_app(app)
@pytest.fixture(scope='session')
def dbconnection(request):
"""Returns an SQLALchemy Connection to the triancore_unittest database.
Starts a transation and creates the tables in it.
The transaction is rolled back in teardown.
"""
def finish():
transaction.rollback()
connection.close()
@agronholm
agronholm / orm.py
Last active December 29, 2015 17:28 — forked from mikeywaites/dates.sql
start_date = arrow.utcnow().replace(months=-2).datetime
end_date = arrow.utcnow().datetime
gen_stmt = db.session.query(func.generate_series(start_date, end_date, timedelta(1)).label('day')).subquery()
donation_day = func.date_trunc(time_unit, Donation.created).label('day')
donation_stmt db.session.query(donation_day, func.sum(Donation.amount).label('amount_raised')).\
filter(created.between(start_date, end_date)).group_by(donation_day).subquery()
query = db.session.query(gen_stmt, coalesce(donation_stmt.c.amount_raised, 0)).outerjoin(donation_stmt, gen_stmt.c.day == donation_stmt.c.day)
def outer(key=None):
def inner():
key = key or ''
print(key)
return inner
outer()()
UnboundLocalError: local variable 'key' referenced before assignment