Skip to content

Instantly share code, notes, and snippets.

class conduit(object):
def __init__(self, iterator):
self.iterator = iterator
def filter(self, predicate):
return conduit(itertools.ifilter(predicate, self.iterator))
def map(self, func):
return conduit(itertools.imap(func, self.iterator))
from django.db import connection
def flatten(*args):
def transform_pipeline(mdl):
data = FlatData()
num_before = len(connection.queries)
for fn in args:
fn(data, mdl)
assert len(connection.queries) == num_before
return data
# computations.py
def add_shareholder_name(shareholder_ix):
def inner(flat, mdl):
flat.shareholder_name = shareholder_ix[mdl.shareholder_id]
return inner
def add_transfered_to_labels(transfered_to_ix):
def inner(flat, mdl):
flat.exercised_from_labels = [
def is_canceled(report_date)
def inner(flat, mdl):
flat.is_canceled = mdl.canceled_date < report_date
return inner
report_date = datetime.today()
flat_securities = (
conduit(all_securities)
.map(
Author
------
authors.id | authors.name
-------------------
1 | "paul"
2 | "peter"
3 | "john"
Books
-----
# Don't waste a query if you are using the queryset
books = Book.objects.filter(..)
if len(books) > 5:
do_stuff_with_books(books)
# If you aren't using the queryset use count
books = Book.objects.filter(..)
if books.count() > 5:
do_some_stuff()
# Don't waste a query if you are using the queryset
books = Book.objects.filter(..)
if books:
do_stuff_with_books(books)
# If you aren't using the queryset use exist
books = Book.objects.filter(..)
if books.exists():
do_some_stuff()
def author_name_length(book):
return len(book.author.name)
def process_author_books(author):
for book in author.books.all():
do_stuff(book)
# Get the Author's name of a Book
book = Book.objects.first()
book.author.name