hartym

def my_transform(i: int) -> str:
    yield str(i)

hartym

def my_load(s: str):
    with open('/tmp/output.txt', 'a+') as f:
        f.write(s+'\n')

hartym

import bonobo

graph = bonobo.Graph(
    my_extract,
    my_transform,
    my_load,
)

if __name__ == '__main__'
    bonobo.run(graph)

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

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
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 bonobo

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

def get_services(**options):
    return {}

hartym

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


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

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.util.bags import BagType
from bonobo_quirksmode.config import set_output_type


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

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