Python
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()