1 - Async Python

gunicorn

< A Python WSGI HTTP server, run on Unix-like OS, inspired by ruby unicorn < pre-fork-worker模式,一个master进程管理多个worker进程 < 推荐的worker数量是:(2 * $num_cores) + 1

pip install gunicorn greenlet eventlet gevent

# -k, --worker-class 工作模式
gunicorn -k sync --workers=17 --threads 1 --worker-connections 1000

sync 多进程模式

一次仅处理一个请求

eventlet, gevent 协程模式

协程实现(cooperative multi-threading),利用非同步IO让一个process在等待IO回应时继续处理下个请求

gthread 多线程模式

线程工作模式,利用线程池管理连接

gaiohttp

利用aiohttp库实现异步I/O

2 - Database client in Python

mysql

# pip3 install PyMySQL
import pymysql

db = pymysql.connect("localhost","user","passwd","testdb" )
cursor = db.cursor()

cursor.execute("SELECT VERSION()")
print(f"Database version : {cursor.fetchone()}")

try:
   cursor.execute("DROP TABLE IF EXISTS EMPLOYEE")
   db.commit()
except:
   db.rollback()

cursor.execute("SELECT 1")
rows = cursor.fetchall()
for row in rows:
    print(row[0])

db.close()

rabbitmq

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1', 5672))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
print("[x] Sent 'Hello World!'")
connection.close()

3 - Install Python

pyenv

curl https://pyenv.run | bash

pyenv install --list
pyenv versions
pyenv global 3.7.15

install python2.7

sudo apt install make gcc g++ patch cmake -y

sudo apt install libssl-dev libbz2-dev libreadline-dev zlib1g.dev libsqlite3-dev libffi-dev lzma-dev libsnappy-dev libjpeg-dev default-libmysqlclient-dev -y
pyenv install 2.7.18

build from source

wget http://www.python.org/ftp/python/3.7.15/Python-3.7.15.tgz
tar -zxvf Python-3.7.15.tgz
cd Python-3.7.15

./configure --prefix=/opt/python3 --enable-optimizations
make
make install
make clean
make distclean

/opt/python3/bin/python3 -V

python2

debian9

# docker pull debian:9
# docker run -d -it --name debian9 debian:9
# https://mirrors.tuna.tsinghua.edu.cn/help/debian/
apt update 
apt install python python-pip -y
pip install -U setuptools
apt install make gcc g++ patch cmake -y
apt install libssl-dev libbz2-dev libreadline-dev zlib1g.dev libsqlite3-dev libffi-dev lzma-dev libsnappy-dev libjpeg-dev default-libmysqlclient-dev -y
# mysql-python
# for old debian: libmysqlclient-dev
apt install default-libmysqlclient-dev -y
pip install mysql-python

# pyarrow
apt install -y -V ca-certificates lsb-release wget
wget https://apache.jfrog.io/artifactory/arrow/$(lsb_release --id --short | tr 'A-Z' 'a-z')/apache-arrow-apt-source-latest-$(lsb_release --codename --short).deb -O /tmp/apache-arrow.deb
apt -y install /tmp/apache-arrow.deb
apt -y update
apt -y install libarrow-dev libarrow-python-dev
rm /tmp/apache-arrow.deb

fedora

# docker pull fedora
# docker run -d -it --name fedora fedora
# https://mirrors.tuna.tsinghua.edu.cn/help/fedora/

# python2 is not included
# dnf whatprovides pip
# sudo alternatives --set python /usr/bin/python2

dnf update && dnf install git curl -y
curl https://pyenv.run | bash
cat <<EOF >> ~/.bashrc
export PYENV_ROOT="\$HOME/.pyenv"
command -v pyenv >/dev/null || export PATH="\$PYENV_ROOT/bin:\$PATH"
eval "\$(pyenv init -)"
EOF
source ~/.bashrc

dnf install make gcc g++ -y
dnf install readline-devel zlib-devel openssl-devel bzip2-devel sqlite-devel libffi-devel lzma-sdk-devel -y
pyenv install 2.7.18
pyenv global 2.7.18
# pyarrow
dnf install cmake libarrow-devel libarrow-python-devel -y
pip install arrow pyarrow

# mysql-python
dnf install mysql-devel -y
pip install mysql-python

4 - IPython

ipython

ipython --pylab

5 - Python Pip

Install pip

wget https://bootstrap.pypa.io/get-pip.py -O - | python

Install packages for Mac ARM

numpy

brew install openblas
OPENBLAS="$(brew --prefix openblas)" pip install numpy

6 - Python Std

base

datetime & time

import datetime
import time

dt = datetime.datetime.now()
unix_sec = time.mktime(dt.timetuple())
dt = datetime.datetime.fromtimestamp(time.time())

s = dt.strftime("%Y-%m-%d %H:%M:%S")
dt = datetime.datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
unix_sec = time.mktime(time.strptime(s, "%Y-%m-%d %H:%M:%S"))
s = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(unix_sec))

random

import random

n_float = random.uniform(0, 10) # [0, 10)
n_float = random.random() # [0, 1.0)

# random.randrange([start], stop[, step])
n_int = random.randint(1, 3) # [1, 3]

s_str = random.choice(['r', 'g', 'b'])
s_list = random.sample(['r', 'g', 'b'], 2)

file

ConfigParser

import ConfigParser
conf = ConfigParser.ConfigParser()
conf.read("myapp.ini")
sections = conf.sections()
section = sections [0]
keys = conf.options("sec")
kvs = conf.items("sec")
val = conf.get("sec", "key")
int_val = conf.getint("sec", "key")

7 - Python Test

unittest

import unittest

class TestStringMethods(unittest.TestCase):

    def test_somecase(self):
        self.assertEqual('foo'.upper(), 'FOO')
        self.assertTrue('FOO'.isupper())
        self.assertFalse('Foo'.isupper())

if __name__ == '__main__':
    unittest.main()
python -m unittest test_module1 test_module2
python -m unittest test_module.TestClass
python -m unittest test_module.TestClass.test_method

python -m unittest -v tests/test_something.py

8 - Python Web

std

python3 -m http.server 3333
python -m SimpleHTTPServer 3333

fastapi

Python 3.7+

http://127.0.0.1:8000/docs for https://github.com/swagger-api/swagger-ui http://127.0.0.1:8000/redoc for https://github.com/Rebilly/ReDoc

pip install "fastapi[all]"
# use the 'app' object in module fastapi-main(or file called fastapi-main.py)
uvicorn fastapi-main:app --reload
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root(secs: float):
    from time import sleep
    sleep(secs)
    return {"message": "Hello World"}

flask

pip install Flask
python -m flask run 
from flask import Flask, request

app = Flask(__name__)

@app.route("/")
def hello_world():
        secs = request.args.get("secs")
        from time import sleep
        sleep(float(secs))
        return {"message": "Hello World"}

if __name__ == '__main__':
    app.run(debug=True)

django

pip install Django
django-admin startproject mysite
cd mysite
python manage.py runserver 0.0.0.0:8000

bottle

from bottle import request, route, run, template
from time import sleep

@route('/hello/<name>')
def hello(name):
    return template('<b>Hello {{name}}</b>!', name=name)

@route('/')
def index():
    secs = request.query.get('secs')
    name = request.query.get('name')
    if secs:
        sleep(float(secs))
    if not name:
        name = "World
    return "Hello {}!".format(name)

run(host='0.0.0.0', port=8000)

trollius

import trollius as asyncio
from trollius import From

@asyncio.coroutine
def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print("Task %s: Compute factorial(%d)..." % (name, i))
        yield From(asyncio.sleep(1))
        f *= i
    print("Task %s completed! factorial(%d) is %d" % (name, number, f))

loop = asyncio.get_event_loop()
tasks = [
    asyncio.async(factorial("A", 8)),
    asyncio.async(factorial("B", 3)),
    asyncio.async(factorial("C", 4))]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()