gevent spawn出來的task,可以用以下三個decorate來綁定, task做完後的callback
glet = gevent.spawn(task, ...)
@glet.link
def on_finish(glet):
pass
# Safe Queue Celery App Settings | |
from __future__ import absolute_import, unicode_literals | |
from celery import Celery | |
app = Celery('some_project', | |
broker='amqp://', | |
backend='amqp://', | |
include=['some_project.tasks']) |
#!/usr/bin/env/python | |
# | |
# More of a reference of using jinaj2 without actual template files. | |
# This is great for a simple output transformation to standard out. | |
# | |
# Of course you will need to "sudo pip install jinja2" first! | |
# | |
# I like to refer to the following to remember how to use jinja2 :) | |
# http://jinja.pocoo.org/docs/templates/ | |
# |
import random | |
import functools | |
import json | |
import weakref | |
import time | |
import sys | |
import gevent | |
class memoize(object): | |
def __init__(self, function): |
# vi: set ts=2 sw=2: | |
# a structure support mongo replica + sharding | |
version: '3.4' | |
services: | |
rs_1: | |
image: mongo:3.2 | |
ports: | |
- 27011:27017 | |
volumes: | |
- ./rs_1.conf:/etc/mongo/mongod.conf:ro |
import functools | |
import logging | |
import inspect | |
invoke_logger = logging.getLogger('invoke_logger') | |
fp = logging.FileHandler('invoke.log') | |
invoke_logger.addHandler(fp) |
import datetime | |
import arrow | |
from flask import Flask, request, current_app | |
from werkzeug.test import EnvironBuilder | |
app = Flask(__name__) | |
@app.route | |
def inedx(): |
from datetime import timedelta | |
import re | |
UNITS = {"s":"seconds", "m":"minutes", "h":"hours", "d":"days", "w":"weeks"} | |
def convert_to_seconds(s): | |
params = {} | |
for amount, u in re.findall(r'(?:(\d+)([smhdw])\s*)', s): | |
u = UNITS[u] | |
amount = int(amount) | |
if u in params: |
import collections | |
import bisect | |
import json | |
import copy | |
import sys | |
Cut = collections.namedtuple('Cut', ['at', 'status']) | |
class Duration(object): | |
def __init__(self, begin_at, end_at): |
gevent spawn出來的task,可以用以下三個decorate來綁定, task做完後的callback
glet = gevent.spawn(task, ...)
@glet.link
def on_finish(glet):
pass
有時候在跑unittest的時候,需要一些前置作業,多半可以寫在setUp或tearDown上
pytest提供fixture, 可以定義每一次的跑unitest的session, module, class, method,都預先跑某個準備函式
例如,我希望跑unittest時先起一個mock database,並在結束時,關掉這個mock database
就可以寫在conftest.py
上
#conftest.py
@pytest.fixture(scope="session", autouse=True)