Skip to content

Instantly share code, notes, and snippets.

View hartym's full-sized avatar
👽
Crafting great software

Romain Dorgueil hartym

👽
Crafting great software
View GitHub Profile
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
hartym / sandbox.sh
Created May 24, 2018 07:22
sandbox.sh
#! /bin/bash
# Exit if something fails
set -e
# Show everything
set -x
# Set version
export VERSION=`git describe`
from contextlib import contextmanager
import bonobo
def get_graph(**options):
graph = bonobo.Graph([1,2,3], bonobo.PrettyPrinter())
return graph
@contextmanager
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'])
{
"name": "my-project",
"dependencies": {
"some-dep": "file:$(python -c 'import os, some_dep; print(os.path.dirname(some_dep.__file__))')/"
}
}
import bonobo
from bonobo.config import ContextProcessor, use_context
from bonobo.constants import NOT_MODIFIED
def extract():
yield 'hello'
yield 'world'
@hartym
hartym / default.py
Created November 29, 2017 10:16
bonobo 0.6 - default.py
import bonobo
def get_graph(**options):
graph = bonobo.Graph()
graph.add_chain(...)
return graph
def get_services(**options):
return {}
@hartym
hartym / example_job.py
Created November 11, 2017 18:43 — forked from cw-andrews/example_job.py
CSV Sanitizer Bonobo ETL Job with Atomic Writes
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
hartym / hello_mondrian.py
Last active November 6, 2017 14:48
Hello, Mondrian
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)
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