Python QA

下面是一份修复后的版本。主要改动:

  • 删除所有失效链接和纯外链引用;
  • 将 Python 2 相关内容更新为 Python 3;
  • 修正原文中容易误导或错误的表述;
  • 代码统一为 Python 3 可运行;
  • 保留原文档结构,但做了精简和勘误。

Python 语言特性(Python 3 修复版)#

1. Python 的函数参数传递#

Python 的参数传递既不是“值传递”,也不是“引用传递”,而是 对象引用传递(call by object reference)。

  • 变量名只是对象的引用;
  • 函数收到的是引用的副本;
  • 如果对象不可变(int、str、tuple、frozenset),函数内重新赋值不会影响外部;
  • 如果对象可变(list、dict、set),函数内原地修改会影响外部。
a = 1
def fun(a):
    a = 2
fun(a)
print(a)  # 1

a = []
def fun(a):
    a.append(1)
fun(a)
print(a)  # [1]

用 id 观察:

a = 1
def fun(a):
    print("func_in", id(a))
    a = 2
    print("re-point", id(a), id(2))
print("func_out", id(a), id(1))
fun(a)
print(a)  # 1

不可变对象重新赋值后,引用指向新对象;可变对象原地修改时,引用不变。

2. Python 中的元类(metaclass)#

元类是“创建类的类”。默认情况下,type 就是所有类的元类。元类常用于 ORM、框架等需要控制类创建过程的场景。日常业务开发很少直接使用。

Python包管理

[[Rye 简单使用指南]]#

Rye 是一个由 Flask 作者 Armin Ronacher 开发的 Python 版本和依赖管理工具,类似于 Pyenv + Poetry 的组合。

安装 Rye#

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh # 安装cargo
cargo install --git https://github.com/mitsuhiko/rye rye # 安装rye

cargo的 .bashrc 相关:

略略略略略略

. "$HOME/.cargo/env"

基本命令#

初始化新项目#

rye init my_project
cd my_project

指定 Python 版本#

rye pin 3.9.10  # 指定使用 Python 3.9.10

添加依赖#

rye add flask==2.3.2
rye add black --dev  # 添加开发依赖

同步依赖#

rye sync

运行项目#

rye run python main.py

常用工具#

rye fmt    # 格式化代码
rye lint   # 代码检查
rye test   # 运行测试

优势特点#

  1. 内置 Python 版本管理,自动下载所需 Python 版本
  2. 使用标准的 pyproject.toml 文件
  3. 无需激活虚拟环境,自动感知项目环境
  4. 内置常用开发工具 (fmt/lint/test)

示例项目结构#

my_project/
├── .python-version
├── pyproject.toml
├── src/
│   └── ...
└── tests/
    └── ...

提示:Rye 适合小型到中型项目,对于大型项目可能需要评估是否满足需求。

Python编码规范

函数返回值定义#

多个返回值 - Tuple#

async def function_name(package_name: str) -> Tuple[ParamsPushLvl2Pool, bool]:
	pass

支持传入多类型 - Union#

ex: Union[ExpiryT, None] = None,

迭代#

def _yield_offer_info(_session, from_days_ago: int) -> Iterator[Offer]:

Python Redis 客户端

异步调用时自动关闭#

当事件循环正在运行时,会创建一个任务来关闭 redis 连接,而不会阻塞事件循环。如果事件循环没有运行,就可以像之前一样调用 run_until_complete。

async def close_redis(self):
    if self.redis:
        await self.redis.close()

def __del__(self):
    loop = asyncio.get_event_loop()
    if loop.is_running():
        asyncio.create_task(self.close_redis())
    else:
        loop.run_until_complete(self.close_redis())

数据迁移(hash数据)#

脚本#

import os
import json
import redis
import argparse
from loguru import logger

home_path = os.environ.get('HOME')

parser = argparse.ArgumentParser(description='Copy Reten Config')
parser.add_argument('--mode', help='Pull or Push data from Redis')
parser.add_argument('-ds', '--datasource', help='retens or events config data from Redis')
args = parser.parse_args()

source_redis = redis.StrictRedis(host='redis-product.com', port=6379, db=0)
def scan_keys_with_pattern(pattern):
    # 初始游标
    cursor = '0'
    count_group = 0
    while True:
        # 使用 SCAN 命令进行迭代
        cursor, keys = source_redis.scan(cursor=cursor, match=pattern)

        # 游标为 '0' 表示迭代结束
        if cursor == b'0' or cursor == 0 or cursor == '0':
            break

        # 将找到的键添加到结果列表中
        count_group += 1
        logger.info(f"{count_group} - {len(keys)} = {cursor}")
        yield [key.decode('utf-8') for key in keys]

def append_2_file(data):
    # 将数据转换为字典
    decoded_data = {}
    for key, value in data.items():
        decoded_data[key.decode('utf-8')] = value.decode('utf-8')
    # 将数据写入文件
    with open(f'data_{args.datasource}.json', 'a') as file:
        file.write(json.dumps(decoded_data) + '\n')

def pull():
    # 连接到源 Redis 实例

    # 从 Redis 中读取数据
    all_keys = set()
    if args.datasource == 'retens':
        for keys in scan_keys_with_pattern('mmp:cfgs:*:retens'):
            for key in keys:
                if key not in all_keys:
                    all_keys.add(key)
                else:
                    logger.warning(f"重复一次: {key}")
                    continue
                data = source_redis.hgetall(key)
                append_2_file(data=data)
    elif args.datasource == 'events':
        for keys in scan_keys_with_pattern('mmp:cfgs:*:events'):
            for key in keys:
                if key not in all_keys:
                    all_keys.add(key)
                else:
                    logger.warning(f"重复一次: {key}")
                    continue
                data = source_redis.hgetall(key)
                append_2_file(data=data)
    else:
        logger.error(f"Invalid datasource: {args.datasource}")
        return

def push():
    # 连接到目标 Redis 实例
    target_redis = redis.StrictRedis(host='localhost', port=6379, db=1)

    # 读取文件内容并写入目标 Redis
    if not args.datasource in ['retens', 'events']:
        logger.error(f"Invalid datasource: {args.datasource}")
        return

    with open(os.path.join(home_path, f'data_{args.datasource}.json'), 'r') as file:
        for l in file:
            l = json.loads(l)
            for key, value in l.items():
                value_obj = json.loads(value)
                pkg_name = value_obj["pkg_name"]
                if args.datasource == 'retens':
                    ok = target_redis.hset(f'mmp:cfgs:{pkg_name}:retens', key, value)
                elif args.datasource == 'events':
                    ok = target_redis.hset(f'mmp:cfgs:{pkg_name}:events', key, value)
                else:
                    print(f"Invalid datasource: {args.datasource}")
                logger.info(f'Writing {key} {value} to Redis -> {ok}')

if __name__ == "__main__":
    if args.mode == 'pull':
        pull()
    elif args.mode == 'push':
        push()

使用#

python {thisscript}.py --mode=pull -ds=events
python {thisscript}.py --mode=push -ds=events

hash#

import redis

# 创建 Redis 连接
r = redis.Redis(host='localhost', port=6379, db=0)

# 写入数据到哈希
r.hset('my_hash', 'field1', 'value1')
r.hset('my_hash', mapping={'field2': 'value2', 'field3': 'value3'})

# 获取单个字段的值
value = r.hget('my_hash', 'field1')
print(f'field1: {value.decode("utf-8")}')

# 获取所有字段和值
all_fields = r.hgetall('my_hash')
print('All fields:')
for field, value in all_fields.items():
    print(f'{field.decode("utf-8")}: {value.decode("utf-8")}')

# 查看哈希中字段的数量
field_count = r.hlen('my_hash')
print(f'Number of fields in my_hash: {field_count}')

事务#

import asyncio
import aioredis

async def run_transaction(redis: aioredis.Redis):
    async with redis.pipeline() as pipe:
        # 开始事务
        await pipe.multi()
        
        # 在事务中执行多个命令
        await pipe.set('key1', 'value1')
        await pipe.set('key2', 'value2')
        
        # 提交事务
        results = await pipe.execute()
        
        print('Transaction results:', results)

async def main():
    # 创建 Redis 连接
    redis = await aioredis.from_url("redis://localhost:6379", encoding="utf-8", decode_responses=True)
    
    # 运行事务
    await run_transaction(redis)
    
    # 关闭连接
    await redis.close()

# 运行异步主函数
if __name__ == "__main__":
    asyncio.run(main())

乐观锁#

详细解释

Python 多线程多进程

  • 一个协程示例:
import asyncio

async def my_coroutine():
    # 这里是协程的代码
    await asyncio.sleep(1)
    print("Coroutine is done")

async def main():
    # 创建一个任务对象
    task = asyncio.create_task(my_coroutine())
    # 等待任务完成
    await task

# 运行主协程
asyncio.run(main())
  • 一个线程示例:
import threading

# 定义一个函数,用于在线程中执行
def thread_function(name):
    print(f"Thread {name}: starting")
    # 在这里执行一些操作
    print(f"Thread {name}: finishing")

# 创建线程的实例
thread = threading.Thread(target=thread_function, args=(1,))
# 启动线程
thread.start()
# 等待线程结束
thread.join()
print("Main thread is done")
  • 一个loop tip:
self.loop = asyncio.get_event_loop()
self.queue.set_loop(loop=self.loop)
await self.af.set_loop(loop=self.loop)
await self.adjust.set_loop(loop=self.loop)

为什么这里使用同一个loop ?
---
在 Python 的 `asyncio` 框架中,事件循环(event loop)是处理异步任务的核心。每个异步应用程序通常只需要一个事件循环,所有的异步任务都应该在这个事件循环中运行。使用同一个事件循环的好处包括:
1. **统一管理**:所有异步任务都在同一个事件循环中运行,这样可以更方便地管理和监控它们。
2. **避免冲突**:如果为不同的任务创建多个事件循环,可能会导致资源竞争或冲突,尤其是在共享资源或执行I/O操作时。
3. **保持上下文一致性**:使用同一个事件循环可以确保所有任务都在相同的上下文中运行,这有助于维护状态的一致性。

在你提供的代码片段中,这样做的目的是确保所有这些组件都在同一个事件循环中运行,以便它们可以正确地协同工作。

如果这些组件是异步的,并且需要与事件循环交互,那么它们需要知道使用哪个事件循环。通过这种方式设置事件循环,可以确保组件在执行异步操作时能够正确地在事件循环中注册和处理。

总之,使用同一个事件循环有助于保持代码的整洁和协调,避免潜在的问题,并确保整个应用程序的异步操作能够顺利地进行。

多进程#

多进程通信打日志#


from multiprocessing import Process, Queue
from loguru import logger
import sys
import time

# 配置日志处理函数
def configure_logger(queue):
    logger.remove()  # 移除默认日志处理
    logger.add(sys.stdout, format="{message}")
    logger.add("logfile.log", format="{message}", rotation="1 MB")
    
    while True:
        try:
            record = queue.get()
            if record is None:
                break  # None 表示日志记录结束
            # 使用 log 方法记录日志
            logger.log(record["level"], record["message"])
        except Exception:
            logger.exception("Failed to process log record")

# 配置工作进程
def worker_process(name, queue):
    logger.remove()  # 移除默认日志处理
    def enqueue_log_message(message):
        # print('message.record["level"].name:', message.record["level"].name)
        # print('message.record["function"]:', message.record["function"])
        # print('message.record["line"]:', message.record["line"])
        # print('message.record.keys:', message.record.keys())
        queue.put({
            "line": message.record["line"], "function": message.record["function"],
            "level": message.record["level"].name, "message": message.strip()
        })
    logger.add(enqueue_log_message)
    
    for i in range(5):
        time.sleep(0.01)
        logger.info(f"Worker {name} is logging iteration {i}")

if __name__ == "__main__":
    log_queue = Queue()
    
    # 启动日志处理进程
    log_process = Process(target=configure_logger, args=(log_queue,))
    log_process.start()
    
    # 启动工作进程
    processes = []
    for i in range(5):
        process = Process(target=worker_process, args=(f"Process-{i}", log_queue))
        processes.append(process)
        process.start()

    # 等待所有工作进程完成
    for process in processes:
        process.join()
    
    # 终止日志处理进程
    log_queue.put(None)
    log_process.join()

解决问题

python

PiP Not Found Issue#

使用pip安装某包时, 提示让更新, 按提示操作更新没效果没反应再用就提示ModuleNotFoundError: No module named 'pip' (ˉ▽ˉ;)…

ModuleNotFoundError: No module named ‘pip’#

升级PiP时出现问题可由下方命令修复#

python -m ensurepip
python -m pip install --upgrade pip

SSL校验#

安装EasyOCR时, reader = easyocr.Reader(['ch_sim','en']) 下载到接近90,结果报错了…. 估计是SSL问题加入以下两条,不知如何.

import ssl
ssl._create_default_https_context = ssl._create_unverified_context

from http.client To revert to the previous, unverified, behavior ssl._create_unverified_context() can be passed to the context parameter.

使用国内镜像下载Python#

	pip install --index https://pypi.mirrors.ustc.edu.cn/simple/ dlib(numpy等包名)
    # 一键更新pip 包资源(利用管道grep传输查询到的需要更新包名,传输到install命令)
    pip3 freeze --local | grep -v '^\-e' | cut -d = -f 1  | xargs -n1 pip3 install -U
    # 权限不够的话就在`pip3 install` 之前加`sudo`反正我不习惯用`root`
安装错误?#

居然有pip3 install XX的错误…这也是因为有旧版pip3存留。需要

Py小工具和功能性方法

author:Ian

python

邮件#

中文附件#

发送邮件时携带附件 中文名称附件

def send_email_week_report(htm, recipients, copys, file_name, file_path):
    port = 465
    # str_today = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    str_today = datetime.now().strftime("%Y-%m-%d")
    mail_host = "mail.{demo}.com"
    mail_user = "{user name}"
    mail_pass = "{passwd}"
    sender = '{name}@{demo}.com'
    mail_from = "{name} <{name}@{demo}.com>"
    msg_Bcc = '{name}@{demo}.com'
    subject = f"{title}-{str_today}"  # 标题
    msg = MIMEMultipart()
    msg.attach(MIMEText(htm, 'html', 'utf-8'))  # 这里可以传html内容 也可以传普通文字

    # 创建附件
    with open(file_path, 'rb') as attachment:
        part = MIMEBase('application', 'octet-stream')
        part.set_payload(attachment.read())
        encoders.encode_base64(part)
        part.add_header(
            'Content-Disposition',
            'attachment',
            filename=("utf-8", "", file_name),
        )
        msg.attach(part)

    for root, _, files in os.walk(os.path.join(BASE_DIR, '{path}/images')):
        for f in files:
            if f in htm:
                with open(os.path.join(root, f), 'rb') as fp:
                    image = MIMEImage(fp.read())
                    image.add_header('Content-ID', f'<{f}>')
                    msg.attach(image)

    msg['From'] = mail_from  # 发件人
    msg['To'] = ";".join(recipients)  # 收件人多个逗号分隔
    msg["Cc"] = ";".join(copys)
    msg["Bcc"] = msg_Bcc
    subject = subject
    msg['Subject'] = Header(subject)
    try:
        smtpObj = smtplib.SMTP_SSL(mail_host)
        if is_partner_node == 1:
            smtpObj = smtplib.SMTP(mail_host)
        smtpObj.connect(mail_host, port)
        smtpObj.login(mail_user, mail_pass)
        smtpObj.sendmail(sender, recipients + copys + [sender], msg.as_string())
        return True
    except smtplib.SMTPException as e:
        return False

有限制的附件#

这种内容插入附件会有文件名称长度限制(19好像或者16)

机器学习库

Python 机器学习库 👽#

Plotly#

与matplotlib 都是绘图工具,不过效果炫一些,我也没画过,所以只放链接,不放实例了 Plotly Python Library : https://plot.ly/python/

matplotlib#

import matplotlib.pyplot as plt

参数等太多,链接最可靠#

pyplot参数

还是粘一些常用的: marker 属性(下面写在分号里呦) o . v ^ < > 1 2 3 4 8 s p * h H + x D d | _ 之类

画出一些“花儿”

绘图#

plt.plot(x, y)
# 在y之后可添加参数,例如常用的label = ‘IamLabel’之类
# 线的样式、颜色 :b: blue  g: green    r: red  c: cyan m: magenta
y: yellow   k: black    w: white
'-' : solid , '--' : dashed, '-.' : dashdot ':' : dotted    '   '', ' '   ': None
# 粗细 lw=3 更改数字
# 数值折点显示样式 marker = ‘o’  

plot.show()

绘制图表#

1#

plt.figure(1)
绘图
plt.figure(2)
绘图

2(未测试)#

plt.figure(1)   # 创建图表1
plt.figure(2)   # 创建图表2
hi1 = plt.subplot(211)  # 在图表2中创建子图1
hi2 = plt.subplot(212)  # 在图表2中创建子图2

一表多图#

pCapital = plt.subplot(4, 1, 1)
    pCapital.set_ylabel("capital")
    pCapital.plot(d['capitalList'], color='r', lw=0.8)
plt.show()

标注点#

eg1#

for w, m in enumerate(self.lowestPrice):
        if w % 120*10 == 0:
            plt.plot([w, w], [m, self.highestPrice[w]], linestyle = '--')
        #plt.scatter(self.dealPoints,color = 'c')
for i in self.dealPoints:  
    plt.scatter([i[0]], [i[1]], color = 'c')  
    for ii in self.ydealPoints:
    plt.scatter([ii[0]], [ii[1]], color = 'm')

plt.title('Tick & TradePoint')
plt.legend()
plt.show()

plt.legend() # show() 之前不加这句是不会显示出标注的呦

Python 常用库

python

pydantic#

来自类#

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

    class Config:
        from_attributes = True

# 假设我们有一个普通的 Python 对象,它具有与模型字段相同的属性名
class OrdinaryObject:
    def __init__(self, name, age):
        self.name = name
        self.age = age

# 创建一个 OrdinaryObject 实例
ordinary_object = OrdinaryObject(name='Alice', age=30)

# 现在我们可以直接将 ordinary_object 传递给 User 模型的构造函数
user_model = User(ordinary_object)

# 输出模型的字段值
print(user_model.dict())  # 输出: {'name': 'Alice', 'age': 30}

自定义字段#

class User(BaseModel):
	id: int
	is_active: bool
	wx_infos: List[str] = []

	class Config:
		arbitrary_types_allowed = True # 允许任意类型

Tuple定义问题#

from typing import List, Tuple
class Task(BaseModel):
	launch_count_today: int
	launch_groups: List[Tuple[str, ...]]
	@property
	def launch_groups_count(self) -> int:
		return len(self.launch_groups)

wrong tuple length, expected 1 (type=value_error.tuple.length; actual_length=5; expected_length=1)#

from typing import List, Tuple
class Task(BaseModel):
	launch_count_today: int
	launch_groups: List[Tuple[str]] # 这样定义就会tuple限制长度
	@property
	def launch_groups_count(self) -> int:
		return len(self.launch_groups)

random#

有权重的随机#

import random
# 假设有一个字典
my_dict = {'a': 1, 'b': 2, 'c': 3}

# 将字典的键和值分别保存在两个列表中
keys = list(my_dict.keys())
weights = list(my_dict.values())

# 使用 random.choices() 函数进行带有权重的随机选择 取12个值
random_choice = random.choices(keys, weights, k=12)

print("随机选择的键:", random_choice)

loguru#

from multiprocessing import Process, Queue
from loguru import logger
import sys
import time

# 配置日志处理函数
def configure_logger(queue):
    logger.remove()  # 移除默认日志处理
    logger.add(sys.stdout, format="{message}")
    logger.add("logfile.log", format="{message}", rotation="1 MB")
    
    while True:
        try:
            record = queue.get()
            if record is None:
                break  # None 表示日志记录结束
            # 使用 log 方法记录日志
            logger.log(record["level"], record["message"])
        except Exception:
            logger.exception("Failed to process log record")

# 配置工作进程
def worker_process(name, queue):
    logger.remove()  # 移除默认日志处理
    def enqueue_log_message(message):
        # print('message.record["level"].name:', message.record["level"].name)
        # print('message.record["function"]:', message.record["function"])
        # print('message.record["line"]:', message.record["line"])
        # print('message.record.keys:', message.record.keys())
        queue.put({
            "line": message.record["line"], "function": message.record["function"],
            "level": message.record["level"].name, "message": message.strip()
        })
    logger.add(enqueue_log_message)
    
    for i in range(5):
        time.sleep(0.01)
        logger.info(f"Worker {name} is logging iteration {i}")

if __name__ == "__main__":
    log_queue = Queue()
    
    # 启动日志处理进程
    log_process = Process(target=configure_logger, args=(log_queue,))
    log_process.start()
    
    # 启动工作进程
    processes = []
    for i in range(5):
        process = Process(target=worker_process, args=(f"Process-{i}", log_queue))
        processes.append(process)
        process.start()

    # 等待所有工作进程完成
    for process in processes:
        process.join()
    
    # 终止日志处理进程
    log_queue.put(None)
    log_process.join()

Python 笔记

python

pip cache 目录#

echo $PIP_CACHE_DIR

特殊函数调用 property#

使用字段的方式调用函数

from datetime import datetime
from pydantic import BaseModel
class InstallCountInfo(BaseModel):
	count: int
	total: int
	@property
	def done_rate(self):
		return round(self.count / self.total, 4)
class InstallCount(BaseModel):
	pass_day: int
	install_day: datetime
	info: InstallCountInfo

彻底摆脱to_dict和from_dict#

使用 pydantic#

BaseModel类型支持:

  • b = BattleAxiePositionInfo.parse_obj(DICT_DATA)
  • b.json()
  • b.dict()
  • parse_file
  • parse_raw
from pydantic import BaseModel
class PositionInfo(BaseModel):
    error: int = -1 # 收集错误
    none: int = 0 # 还没开始
    clicked: int = 1 # 在client 赋此值
    done: int = 2 # 在server 赋此值
    xy: List[int] = [0, 0]
    status: int = 0 # clicked or done or none or error
class BattleAxiePositionInfo(BaseModel):
    our: List[PositionInfo] = [PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo()]
    enemy: List[PositionInfo] = [PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo()]

pp = BattleAxiePositionInfo()
print(f"pp json: {pp.json()}")
dict_pp = pp.dict()
pp = BattleAxiePositionInfo.parse_obj(dict_pp)

pp json: {“our”: [{“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}], “enemy”: [{“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}, {“error”: -1, “none”: 0, “clicked”: 1, “done”: 2, “xy”: [0, 0], “status”: 0}]}