Skip to content

Instantly share code, notes, and snippets.

@allenyang79
allenyang79 / celery.py
Created June 18, 2018 10:18 — forked from IrSent/celery.py
Celery Best Practices
# 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'])
@allenyang79
allenyang79 / jinja2_file_less.py
Created June 8, 2018 03:34 — forked from wrunk/jinja2_file_less.py
python jinja2 examples
#!/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/
#
@allenyang79
allenyang79 / memorize_instance_method.py
Last active May 31, 2018 07:55
memorize_instance_method.py
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
@allenyang79
allenyang79 / invoke_wrap.py
Last active May 8, 2018 08:43
trace which invoke current function
import functools
import logging
import inspect
invoke_logger = logging.getLogger('invoke_logger')
fp = logging.FileHandler('invoke.log')
invoke_logger.addHandler(fp)
@allenyang79
allenyang79 / flask-context.py
Last active May 7, 2018 02:40
flask custom a context
import datetime
import arrow
from flask import Flask, request, current_app
from werkzeug.test import EnvironBuilder
app = Flask(__name__)
@app.route
def inedx():
@allenyang79
allenyang79 / convert_to_seconds.py
Created April 26, 2018 06:15
process time period string like 'w', 'd', 'h', 'm' , 's'
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:
@allenyang79
allenyang79 / duration-0.py
Last active April 12, 2018 06:01
duration.py
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
 
@allenyang79
allenyang79 / 0_pytest.md
Last active March 21, 2018 06:37
@pytest.fixture(scope="session", autouse=True)

有時候在跑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)