ficapy

import requests
import inspect
from collections import defaultdict


def _tree(): return defaultdict(_tree)


tree = _tree()

ficapy

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: Ficapy
# Create: '8/22/16'

import requests
import ipaddress
import asyncio
import aiohttp

ficapy

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: Ficapy
# Create: '16/2/23'
import sys

if sys.version_info[0] < 3:
    sys.exit(1)

from github import Github

ficapy

1. 启动celery

   celery -A base_celery worker --loglevel=info 
   

2. 关闭celery(所有任务将丢失不会重启)

pkill -9 -f 'celery worker'

ficapy

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: Ficapy
# Create: '15/10/27'

from pyquery import PyQuery as pq
from operator import attrgetter
from concurrent.futures import ThreadPoolExecutor, as_completed
from operator import methodcaller
import requests

ficapy

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: Ficapy
# Create: '12/14/16'

import pytest
from sqlalchemy import create_engine, Column, BIGINT
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.exc import IntegrityError

ficapy

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: ficapy
# Create: '5/25/17'

import inspect
from datetime import datetime
from sqlalchemy import inspection
from collections import namedtuple
from sqlalchemy import create_engine, Column, Integer, String, DateTime

ficapy

# !/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: ficapy

import time
import threading
from threading import Thread
from sqlalchemy import create_engine, Column, Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

ficapy

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: ficapy
import threading
import functools
import tornado.ioloop
import tornado.web
from sqlalchemy import Column, create_engine
from sqlalchemy.dialects.mysql import INTEGER, BIT, VARCHAR
from sqlalchemy.ext.declarative import declarative_base

ficapy

#https://gist.github.com/gregburek/1441055
#限制单位时间内函数执行次数
import time
from collections import deque


def RateLimited(minutes=1, limitNum=3):
    status = deque([0] * limitNum, maxlen=limitNum)
    def decorate(func):
        def wrap(*args, **kwargs):