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)