python pip

性能调试

最佳线程数目 = （（线程等待时间+线程CPU时间）/线程CPU时间）* CPU数目原文链接：https://cloud.tencent.com/developer/article/1736560

用 APM 工具：

python -m cProfile dev.py
py-spy top --pid 12345
py-spy top -- python3 rys_sitemap.py

数据结构

number

int(x) #将x转换为一个整数。
float(x) #将x转换到一个浮点数。
complex(x) #将x转换到一个复数，实数部分为 x，虚数部分为 0。
complex(x, y) #将 x 和 y 转换到一个复数，实数部分为 x，虚数部分为 y。x 和 y 是数字表达式。

list 列表 []
tup 元组（）元组的元素不能更改
dict 字典 {} { key1 : value1, key2 : value2 }
set 集合 {value01,value02,...} 或 set( )
迭代器和生成器（generator）

迭代器有两个基本的方法：iter() 和 next()。

>>> list=[1,2,3,4]
>>> it = iter(list)   # 创建迭代器对象
>>> next(it)  # 输出迭代器的下一个元素

使用了 yield 的函数被称为生成器（generator）

跟普通函数不同的是，生成器是一个返回迭代器的函数，只能用于迭代操作，更简单点理解生成器就是一个迭代器。

在调用生成器运行的过程中，每次遇到 yield 时函数会暂停并保存当前所有的运行信息，返回 yield 的值, 并在下一次执行 next() 方法时从 yield 的下一条语句继续执行.

#!/usr/bin/python
# -*- coding: UTF-8 -*-
 
def fab(max): 
    n, a, b = 0, 0, 1 
    while n < max: 
        yield b      # 使用 yield
        # print b 
        a, b = b, a + b 
        n = n + 1
 
for n in fab(5): 
    print n

cookiecutter(项目管理工具)

https://github.com/cookiecutter/cookiecutter

conda

官方下载 streamlit

brew install --cask anaconda
PREFIX=/opt/homebrew/anaconda3
echo 'export PATH=/opt/homebrew/anaconda3/bin:$PATH' >> ~/.zshrc
source ~/.zshrc
# or 
echo 'export PATH=/usr/local/anaconda3/bin:$PATH' >> ~/.bash_profile
source ~/.bash_profile 

conda --version
conda create -n stenv python=3.9
conda activate stenv
python --version
conda deactivate # 退出环境
conda env remove -n stenv  # 删除环境，
pip install streamlit

Pyenv (python 版本管理器)

conda

pyenv 自动安装

https://github.com/pyenv/pyenv/blob/master/COMMANDS.md

常见问题

profile、bash_profile、bashrc、区别

echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.bashrc
echo 'export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.bashrc
echo 'eval "$(pyenv init --path)"' >> ~/.bashrc
source ~/.bashrc

# 常见问题
yum install gcc zlib-devel bzip2 bzip2-devel readline-devel sqlite sqlite-devel openssl-devel tk-devel libffi-devel xz-devel -y

------

pyenv update # 升级
pyenv local --unset  # 取消设置local版本：pyenv local --unset
pyenv global
pyenv shell
pyenv install --list
pyenv uninstall
pyenv rehash
pyenv version
pyenv versions
pyenv which
pyenv whence

venv

# 在当前目录创建虚拟环境
python -m venv .
python -m venv .venv
source .venv/bin/activate
deactivate

Pipenv (pip + virtualenv 依赖包管理器) github:https://github.com/pypa/pipenv 文档：https://pipenv.kennethreitz.org/en/latest/

1、安装pipenv 2、 cd project_path & pipenv --python python3

# https://pypi.org/project/pipenv/
pipenv --python 3 #x.x.x版本的创建虚拟环境--初始化项目
pipenv shell #(激活虚拟环境)进入虚拟环境命令行
activate #停止虚拟环境
pipenv --rm #删除虚拟环境
pipenv install --pypi-mirror https://pypi.tuna.tsinghua.edu.cn/simple  #安装软件包 https://pypi.tuna.tsinghua.edu.cn/simple/
pipenv install --dev pytest  #安装 pytest 到开发环境（开发包 Pipfile:[dev-packages]）
pipenv install --dev #依赖包，开发包都安装
pipenv update #更新所有依赖包
pipenv --venv #查看虚拟环境路径
pipenv --where #找到项目
pipenv --py #获取虚拟环境python解释器路径
pipenv graph #查看依赖关系
pipenv sync #已安装的依赖包
pipenv run python Main.py #使用Pipenv虚拟环境运行Main.py
pip freeze > requirements.txt # 生成requirements.txt文件
pipenv run pip freeze > requirements.txt # 生成requirements.txt文件

pip

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

pip freeze > requirements.txt

# 清空安装包
pip freeze > installed.txt
pip uninstall -r installed.txt -y

poetry + pyenv

# 在项目的根目录下执行以下命令，会把虚拟环境创建在 .venv
pyenv local 3.10.6
poetry config virtualenvs.in-project true  # 设置创建虚拟环境到当前目录的.venv
poetry env use python  # 创建虚拟环境，会生成.venv

python 软件目录结构规范

目录结构规范 python风格规范

AI—translate/
|-- bin/
|   |-- dev.py
|
|-- app/
|   |-- tests/
|   |   |-- __init__.py
|   |   |-- test_main.py
|   |-- env.py
|   |-- __init__.py
|   |-- main.py
|
|-- docs/
|   |-- conf.py
|   |-- abc.rst
|
|-- setup.py
|-- requirements.txt
|-- README


# autoload.py 引入模块规范
# 导入应该按照从最通用到最不通用的顺序分组
# 1.标准库导入
import time
import datetime
import requests
import json

# 2.第三方库导入
from loguru import logger

# 3.本地代码子包导入
from app.env import *
from app.database.es_db import *
from app.database.redis_db import *

# 每种分组中, 应该根据每个模块的完整包路径按字典序排序, 忽略大小写.

简要解释一下:
bin/: 存放项目的一些可执行文件，当然你可以起名script/之类的也行。
foo/: 存放项目的所有源代码。(1) 源代码中的所有模块、包都应该放在此目录。不要置于顶层目录。(2) 其子目录tests/存放单元测试代码； (3) 程序的入口最好命名为main.py。
docs/: 存放一些文档。
setup.py: 安装、部署、打包的脚本。
requirements.txt: 存放软件依赖的外部Python包列表。
README: 项目说明文件。
规范：文件名，项目名用下划线，方法名可以用驼峰，因为编辑器默认不区分大小写，文件名用驼峰会冲突。

ORM

SQLAlchemy flask-sqlalchemy SQLAlchemy Session机制研究

peewee

from sqlalchemy import *
from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base

engine = create_engine(f"mysql+pymysql://{mysql_user}:{mysql_password}@{mysql_host}/{mysql_dbname}?charset=utf8",
                       poolclass=sqlalchemy.pool.NullPool)
# 如果设置poolclass = NullPool，engine将不会再使用连接池，那么连接将会在 session.close()后直接关闭.
session_factory = sessionmaker(bind=engine)     # sessionmaker本质上实现了一个工厂模式
session = session_factory() # 通过这个工厂方法来创建一个Session实例


‘’‘并发场景下推介使用线程安全的scoped_session‘’‘
session_factory = scoped_session(sessionmaker(bind=engine))
# scoped_session生成session实例时，会把session保存到一个注册表中，而这个注册表是一个通过threading.local()生成的Thread Local对象，虽然这个对象是全局的，但这个全局变量只有在当前线程才能访问
session.close() # 会将连接返回给Engine的连接池，并且不会关闭连接。

engine.dispose() # 将关闭连接池的所有连接.

peewee

mysql_db = MySQLDatabase('my_database')

class BaseModel(Model):
    """A base model that will use our MySQL database"""
    class Meta:
        database = mysql_db

class User(BaseModel):
    username = CharField()
    # etc, etc

api

fastapi

pip install fastapi # 安装
pip install uvicorn # 安装uvicorn来作为服务器
uvicorn main:app --reload # dev 环境
nohup uvicorn main:app --host 0.0.0.0 --reload --workers 9 >> logs/fastapi.log 2>&1 &
nohup uvicorn main:app --host 0.0.0.0 --reload --workers 17 > /dev/null 2>&1 &
nohup gunicorn main:app -b 0.0.0.0:8000 -w 4 -k uvicorn.workers.UvicornH11Worker --daemon > logs/fastapi.log 2>&1 &
gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000  # 开发部署

ps -ef | grep "python dev.py" | grep -v grep | awk '{print $2}' | xargs kill -9

Flask DOC

# 安装
pip install Flask
export FLASK_ENV=development # 开发环境
flask run
flask run -h 0.0.0.0
env FLASK_APP=app.py flask run --host=127.0.0.1 --port=8080   # 前台启动
(env FLASK_APP=app.py flask run --host=127.0.0.1 --port=8080 > /dev/null &)
(flask run >> logs/linkedin_api.log 2>&1 &)
pipenv run flask run -h 127.0.0.1 -p 8080 # 前台启动
pipenv run flask run -h 0.0.0.0 -p 8080
nohup pipenv run flask run -h 127.0.0.1 -p 8080& #后台启动

env FLASK_APP=app.py FLASK_DEBUG=1 flask run --host=127.0.0.1 --port=8080
(env FLASK_APP=app.py flask run --host=0.0.0.0 --port=7777 > /dev/null 2>&1 &) 
(flask run --host=0.0.0.0 --port=7777 >> logs/translateapi.log 2>&1 &)
export FLASK_APP=app.py
 # 要启用所有开发功能（包括调试模式），可以 在运行服务器之前导出FLASK_ENV环境变量并将其设置为development：
export FLASK_ENV=development

故障排查

性能监控p y-spy

python log

from cmreslogging.handlers import CMRESHandler
from app.vendor.func import es_host, eslog_index
from loguru import logger

handler = CMRESHandler(hosts=[es_host], auth_type=CMRESHandler.AuthType.NO_AUTH, es_index_name=eslog_index)
logger.add(handler)

filename = os.path.splitext(os.path.basename(__file__))[0]
logger.add('%s/logs/%s_{time}.log' % (root_path, filename), rotation='00:00', retention='7 days', enqueue=True)

pool

import concurrent.futures
import time

def run_pool():
    time.sleep(1)

with concurrent.futures.ThreadPoolExecutor() as pool:
    p1 = pool.submit(run_pool)  # 异步提交
    p2 = pool.submit(run_pool)
    p3 = pool.submit(run_pool)

    pool.shutdown(wait=True)  # 等待线程池所有任务结束