jomido

Do: pip install -r requirements.txt Do: python taskqueue.py

You will need a service file. Amend the location in the entry point in taskqueue.py.

jomido

import asyncio
import time

from pools import threads, processes

in_thread = threads(10)
in_process = processes(4)


def fib(n):

jomido

from functools import partial as part


# some utils


def partial(func, *args, **kwargs):

    # replace partial

jomido

partial application

what

Partial application lets you lock in one or more arguments to a function. It gives you a new function that only takes the remaining, as-yet-to-be-specified arguments. If arguments are applied against a function, then partial application allows you to apply only some of them - that is, you can partially apply them. Later on, you can finish out the argument application.

One can also read the Python docs.

from functools import partial

jomido

from functools32 import lru_cache
from functools import partial

memoize = partial(lru_cache, maxsize=128)

@memoize()
def function_to_memoize(arg1, arg2, kwarg1=None):
    pass

jomido

"""
An asynchronous queue for Google Appengine Task Queues

For `auth` and `http_tools` imports, see:

  https://gist.github.com/jomido/93940858a803327197314ceae8b31462
"""

import asyncio
import base64

jomido

"""
Google Cloud auth via service account file
"""

# stdlib
import datetime
import time
import typing

# 3rd party

jomido

always = lambda f: True


def get_filenames(root_path, max_depth=1, predicate=None):

    predicate = predicate or always
    path = os.path.normpath(root_path)
    filenames = []

jomido

import asyncio


class Counter(object):

    def __init__(self, max=500, duration=100, loop=None):

        self.max = max
        self.duration = duration
        self.loop = loop or asyncio.get_event_loop()

jomido

import asyncio
import aiohttp
import time

from utils import start, auto


class Context(object):