inactivist

First steps with the Luigi workflow manager

As an introduction into Luigi, I am following this tutorial with some modifications, e.g. installation using conda.

The problems and solutions described in the examples below have led to the development of sciluigi,

inactivist

from collections import defaultdict, OrderedDict

import luigi
from luigi.task import flatten, getpaths


def topological_sorting(struct, outnodes_funct, transform_funct):
    struct = flatten(struct.keys()) if isinstance(struct, dict) else flatten(struct)
    visited = OrderedDict()
    def dvisit(root):

inactivist

Populate an arq.connections.RedisSettings instance from a redis: connection URL string.

Useful for populating arq WorkerSettings etc.

inactivist

def run_all_tests():
    # pytest would be nicer
    import types

    for name, member in globals().items():  # NB: not iteritems()
        if isinstance(member, types.FunctionType) and name.startswith("test_"):
            print("Run test:", member.__name__)
            member()

if __name__ == "__main__":

inactivist

dwc_otg.lpm_enable=0 console=serial0,115200 console=tty1 root=PARTUUID=xxxxxx rootfstype=ext4 elevator=deadline fsck.repair=yes rootwait quiet splash plymouth.ignore-serial-consoles

inactivist

# see rkengler.com for related blog post
# https://www.rkengler.com/how-to-capture-network-traffic-when-scraping-with-selenium-and-python/

import json
import pprint

from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities

capabilities = DesiredCapabilities.CHROME