hartym

from collections import deque

import bonobo
from bonobo.config import use_context_processor


def rolling_deque(self, context):
    buffer = yield deque()
    while len(buffer):
        retval = buffer.popleft()

hartym

#! /bin/bash

# Exit if something fails
set -e

# Show everything
set -x

# Set version
export VERSION=`git describe`

hartym

from contextlib import contextmanager
import bonobo


def get_graph(**options):
    graph = bonobo.Graph([1,2,3], bonobo.PrettyPrinter())
    return graph


@contextmanager

hartym

import bonobo
from bonobo.util.bags import BagType
from bonobo_quirksmode.config import set_output_type


def generate():
    yield from ('a', 'b', 'c')

LetterBag = BagType('LetterBag', ['letter'])

hartym

{
  "name": "my-project",
  "dependencies": {
    "some-dep": "file:$(python -c 'import os, some_dep; print(os.path.dirname(some_dep.__file__))')/"
  }
}

hartym

import bonobo
from bonobo.config import ContextProcessor, use_context
from bonobo.constants import NOT_MODIFIED


def extract():
    yield 'hello'
    yield 'world'

hartym

import bonobo

def get_graph(**options):
    graph = bonobo.Graph()
    graph.add_chain(...)
    return graph

def get_services(**options):
    return {}

hartym

import argparse
from time import sleep

import os
import re

import bonobo
from atomicwrites import atomic_write
from fs.osfs import OSFS
from texools.files import new_file_check

hartym

import logging
import mondrian

# One line setup (excepthook=True tells mondrian to handle uncaught exceptions)
mondrian.setup(excepthook=True)

# Use logging, as usual.
logger = logging.getLogger()
logger.setLevel(logging.INFO)

hartym

import argparse
import datetime
import os
import time

import bonobo
from atomicwrites import atomic_write
from bonobo.constants import NOT_MODIFIED
from fs.osfs import OSFS