devStandard/docs/learning/9-python/3-高级.md

python高级

闭包

函数1嵌套函数2, 函数1返回函数2, 函数2使用了外部的变量 用处: 防止全局变量被外部修改

def outer_function(x):
    def inner_function(y):
        return x + y
    return inner_function

# 创建一个闭包
closure = outer_function(10)

# 闭包被使用
result = closure(5)
print(result)  # 输出 15

• 如何要修改外部函数变量的值, 需要用nonlocal修饰

def outer_function():
    x = 10  # 外部函数中定义的变量

    def inner_function():
        nonlocal x  # 使用 nonlocal 关键字声明 x 是外部函数中的变量
        x += 5
        print("内部函数中修改后的 x:", x)

    inner_function()
    print("外部函数中的 x:", x)

# 调用外部函数
outer_function()

其他: global修饰

global_variable = 10  # 全局变量

def modify_global_variable():
    global global_variable  # 使用 global 关键字声明在函数内部使用全局变量
    global_variable += 5

modify_global_variable()
print(global_variable)  # 输出 15，因为函数内部修改了全局变量的值

装饰器

在不破坏目标函数的功能前提下, 为目标函数添加功能 实际就是用闭包实现的

def my_decorator(func):
    def wrapper():
        print("在调用被装饰函数之前的操作")
        func()  # 调用原始函数
        print("在调用被装饰函数之后的操作")

    return wrapper  # 返回新的函数

@my_decorator  # 使用装饰器语法
def say_hello():
    print("Hello!")

# 调用被装饰后的函数
say_hello()

其实是语法糖

def my_decorator(func):
    def wrapper():
        print("在调用被装饰函数之前的操作")
        func()  # 调用原始函数
        print("在调用被装饰函数之后的操作")

    return wrapper  # 返回新的函数


def say_hello():
    print("Hello!")

fn = my_decorator(say_hello)
fn()

• 把字典转化成对象

from dataclasses import dataclass

@dataclass
class Torrent:
    name: str
    size: int

# 定义一个字典
torrent_dict = {
    'name': 'example_torrent',
    'size': 1024
}

# 使用字典初始化对象
torrent_obj = Torrent(**torrent_dict)

# 现在你可以访问对象的属性
print(torrent_obj.name)
print(torrent_obj.size)

设计模式

单例, 工厂, 建造者, 责任链, 状态, 备忘录, 解释器, 访问者, 观察者, 中介, 模板, 代理模式

单例模式

某些场景下，我们需要一个类无论获取多少次类对象，都仅仅提供一个具体的实例, 以节省创建类对象的开销和内存开销 定义: 保证一个类只有一个实例,并提供一个访问它的全局访问点

简单案例:

class Tools:
  pass

str_tools = Tools()

from test import str_tools
s1 = str_tools()
s2 = str_tools()
print(s1)
print(s2)
# s1 和 s2 是同一个对象

案例2:

def singleton(cls):
    instances = {}  # 存储实例的字典

    def get_instance(*args, **kwargs):
        if cls not in instances:
            # 如果实例不存在，创建一个新的实例
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance

@singleton
class MyClass:
    def __init__(self, name):
        self.name = name

# 创建两个实例，实际上是同一个实例
obj1 = MyClass("Instance 1")
obj2 = MyClass("Instance 2")

# 输出实例的名称
print(obj1.name)  # 输出 "Instance 1"
print(obj2.name)  # 输出 "Instance 1"，因为实际上是同一个实例

工厂模式

大量创建对象时, 提供一个统一的入口 当发生变化的时候, 仅修改工厂类的创建方法

class Dog:
    def speak(self):
        return "汪汪汪!"

class Cat:
    def speak(self):
        return "喵喵喵!"

class AnimalFactory:
    def create_animal(self, animal_type):
        if animal_type == "Dog":
            return Dog()
        elif animal_type == "Cat":
            return Cat()
        else:
            raise ValueError("不支持的动物类型")

# 使用工厂创建对象
factory = AnimalFactory()

dog = factory.create_animal("Dog")
cat = factory.create_animal("Cat")

# 调用对象的方法
print(dog.speak())  # 输出 "汪汪汪!"
print(cat.speak())  # 输出 "喵喵喵!"

多线程

import threading

thread_obj = threading.Thread([group [, target [, name [, args [, kwargs]]]]])
# group 暂时无用, 未来功能的预留参数
# target 执行的任务目标名
# args 以元组方式传参
# kwargs 以字典方式传参
# name 线程名, 一般不用设置

thread_obj.start()

socket

需要服务端+客户端

服务端

# 服务端
# 1. 创建socket对象
import socket

socket_server = socket.socket()

# 2. 服务端绑定ip和端口
socket_server.bind(('0.0.0.0', 8888))

# 3. 监听端口, 括号内为连接数量, 不填会自动设置一个合理的值
socket_server.listen(5)

# 4. 接受客户端连接
client, addr = socket_server.accept()
print('连接地址:', addr)
# accept()方法会阻塞, 直到有客户端连接才会往下执行Í
# accept()返回二元元组, 可以用两个变量接收

# 5. recv接收
while True:
  data = client.recv(1024).decode("UTF-8")
  # recv()参数为缓存区大小, 设置为1024, 代表缓存区为1024字节
  # recv()返回值是字节数组(bytes对象), 需要decode()方法将字节数组转为字符串
  print(f"接收的数据: {data}")
  if data == "exit":
    break
  # 6. send发送数据
  client.send(f"已接收数据: {data}".encode("UTF-8"))

# 7. 关闭连接
client.close()
socket_server.close()

# 客户端可以先通过 https://github.com/nicedayzhu/netAssist/releases 测试

客户端

# 1. 创建服务
import socket
socket_client = socket.socket()

# 2.建立连接
socket_client.connect(('127.0.0.1', 8888))

while True:
  msg = input("请输入要发送的数据: ")
  # 3. 发消息
  socket_client.send(msg.encode("UTF-8"))
  if msg == "exit":
    break
  # 4. 接收消息
  # recv 是阻塞式的
  data = socket_client.recv(1024).decode("UTF-8")
  print(f"接收的数据: {data}")

# 5. 关闭
socket_client.close()

===服务端先接收再发送, 服务端先发送再接收===

异步

httpx

asyncio

with as

with <expression> as <variable>:
    # 代码块

• <expression>：通常是一个上下文管理器对象，实现了__enter__和__exit__方法。
• <variable>：可选，用于存储由上下文管理器返回的对象。

with语句在进入和退出代码块时分别调用对象的__enter__和__exit__方法，这样可以确保资源在合适的时候被正确获取和释放。例如，在处理文件时：

with open('file.txt', 'r') as file:
    content = file.read()
    # 在这里使用文件内容进行操作
# 在此处退出上下文，文件会被正确关闭

列表推导式

[expression for item in iterable if condition]

• expression：对可迭代对象中每个元素进行操作或处理的表达式。
• item：可迭代对象中的每个元素。
• iterable：用于迭代的对象，如列表、元组、集合等。
• condition：可选的条件，用于过滤元素。如果条件为真，元素将包含在最终的列表中

设置pip 镜像

mkdir ~/.pip
vim ~/.pip/pip.conf

===
[global]
trusted-host =  mirrors.aliyun.com
index-url = http://mirrors.aliyun.com/pypi/simple
===

生成requirements.txt

pip install pipreqs
pipreqs .

引用

-r demo/requirements.txt

版本管理和虚拟环境

pyenv 版本管理

brew install pyenv

# homebrew安装的python环境变量可以屏蔽了
# 比教程多了 $PYENV_ROOT/shims

echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.zshrc
echo '[[ -d $PYENV_ROOT/bin ]] && export PATH="$PYENV_ROOT/bin:$PYENV_ROOT/shims:$PATH"' >> ~/.zshrc
echo 'eval "$(pyenv init -)"' >> ~/.zshrc

pyenv install -l
pyenv install 3.10.4

pyenv uninstall <versions>

pyenv global 3.10 # 设置全局
pyenv global # 查看全局

pyenv shell <version> # 设置当前shell

pyenv local <version> # 设置当前目录 生成 .python-version 文件
pyenv local # 查看当前目录, 读取 .python-version 文件

# 查看已安装的版本
pyenv versions
# 查看当前目录的版本
pyenv version

venv 自带虚拟环境

venv 是python 3.3后官方自带的, 在这之前就有第三方virtualenv

缺点: 只能使用当前系统已经安装好的python，无法使用其他版本的python环境

# 添加虚拟环境
python3 -m venv .venv
# 激活
source .venv/bin/activate
# 正常使用pip
pip install -r requirements.txt
# 取消激活
deactivate

• vscode 自动激活

"python.terminal.activateEnvInCurrentTerminal": true

cmd+shift+p : python create env

看vscode下方状态栏是否显示env和python版本

• zsh 钩子自动激活

vim ~/.zshrc

# 定义 chpwd 钩子
chpwd_hook() {
    # 检查当前目录中是否存在 venv 或 .venv 目录
    if [ -d "venv" ]; then
        # 如果存在 venv 目录，则激活虚拟环境
        source venv/bin/activate
        echo '激活python环境'
    elif [ -d ".venv" ]; then
        # 如果存在 .venv 目录，则激活虚拟环境
        source .venv/bin/activate
        echo '激活python环境'
    fi
}

# 将 chpwd_hook 设置为 chpwd 钩子
chpwd_functions+=(chpwd_hook)

• direnv

https://direnv.net/docs/hook.html

brew install direnv

# 添加到~/.zshrc
eval "$(direnv hook zsh)"
# 关闭提示
export DIRENV_LOG_FORMAT=

# 在项目下新建.envrc
echo 'source $PWD/.venv/bin/activate' > .envrc

# 允许当项目下加载配置
direnv allow

常见函数:

https://direnv.net/man/direnv-stdlib.1.html

PATH_add bin

uv 包管理器

用来代替 pip pip-tools virtualenv 命令 有全局缓存

brew install uv

# 创建虚拟环境
uv venv
source .venv/bin/activate

# 或者不激活直接 uv run xx.py

uv pip install xxx
uv pip install -r requirements.txt
uv pip sync requirements.txt

# 生成依赖列表
uv pip freeze > requirements.txt

# 指定系统中某个版本的python
PYENV_VERSION=3.12.2 uv venv
--python 参数可以指定pyton路径

• 设置镜像

export UV_DEFAULT_INDEX="http://mirrors.aliyun.com/pypi/simple"

• macos 使用硬链接

export UV_LINK_MODE="hardlink"

rye 虚拟环境和包管理

包管理可以使用 uv

miniconda/miniforge

适合公司使用, vscode不会自动切换环境

miniforge是社区免费版, miniconda是公司产品

brew install miniforge

# 设置镜像源
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/pytorch/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/
conda config --set show_channel_urls yes
conda config --show channels
conda update --all

# 默认创建了一个base的环境

# 用Python 3.9创建一个名为 f0x 的虚拟环境
conda create -n f0x python=3.9

# 查看所有环境
conda info --envs
conda env list 

# 激活虚拟环境
conda activate f0x

# 安装库
conda install jupyter

poetry 虚拟环境和包管理

https://blog.kyomind.tw/python-poetry/

https://python-poetry.org/docs/main/configuration/

curl -sSL https://install.python-poetry.org | python3 -
# 安装路径 $HOME/.local/bin
# poetry 要安装在单独的虚拟环境中, 因为它自己就带一堆包, 并且会依赖特定版本的python
# 卸载
curl -sSL https://install.python-poetry.org | POETRY_UNINSTALL=1 python3 -

vim ~/.zshrc
===
alias p="poetry"
===
source ~/.zshrc

# 初始化
poetry init

# 安装环境当前工作目录下
poetry config virtualenvs.in-project true
# 查看配置
poetry config --list

poetry env use python
# poetry env use python3
# poetry env use python3.8
# poetry env use 3.8
poetry env remove python

# 进入虚拟环境
poetry shell
# 退出
exit

# 相当于pip list
poetry show
poetry show --tree
poetry show 模块名 --tree

# 不进入shell也可以安装
poetry add 模块名
poetry remove 模块名
# 安装到开发环境
poetry add 模块名 --group dev

# 更新所有模块
poetry update
# 更新指定模块
poetry update 模块名

# 手动修改pyproject.toml后不会自动更新lock, 需要执行
poetry lock
poetry install --sync

# 导入requirements.txt 
cat requirements.txt | xargs poetry add

# 导出requirements
poetry export -f requirements.txt -o requirements.txt --without-hashes

# 按照 pyproject.toml 安装
poetry install

# 不进入环境执行运行
poetry run python3 main.py

# 更多指令
poetry list

# vscode 使用
poetry env info
poetry env info --path
# 手动指定python路径

json

# 内置模块 json
import json

data = [{"name":"张三", "age":18, "gender":"男"}]

# py数据转json
data = json.dumps(data)
print(data)
print(type(data))

# 中文变乱码
data = json.dumps(data, ensure_ascii=False)
print(data)
print(type(data))

# json转py数据
data = json.loads(data)
print(data)
print(type(data))

三目运算符

x if condition else y

# 示例

x = 5
result = "Even" if x % 2 == 0 else "Odd"
print(result)

逻辑短路

# and 运算符的逻辑短路
result_and = False and some_function()  # some_function() 不会被调用

# or 运算符的逻辑短路
result_or = True or some_function()  # some_function() 不会被调用

pycache

python编译成字节码的产物, 直接vscode和gitignore忽略掉

锁

import threading

# 创建一个锁
lock = threading.Lock()

# 使用锁来保护共享资源
shared_resource = 0

def update_shared_resource():
    global shared_resource
    with lock:
        shared_resource += 1

# 创建多个线程来同时更新共享资源
threads = []
for _ in range(5):
    t = threading.Thread(target=update_shared_resource)
    threads.append(t)
    t.start()

# 等待所有线程执行完毕
for t in threads:
    t.join()

print(shared_resource)  # 应该输出 5

• 装饰器写法

def locked(func):
    def wrapper(*args, **kwargs):
        with lock:
            return func(*args, **kwargs)

    return wrapper

代码检查工具

• ruff

ruff src
# 递归检查
ruff src -- --recursive
# 修复
ruff src -- --recursive --fix

vscode 有 ruff 插件

• blank

black --check src
black src

测试

• pytest

# 在src目录下创建test目录
# 在test目录下创建 __init__.py
# 测试脚本文件名  test_xxx.py, test_开头
from module.xxxx import  xxxxx

def test_xxx():
	# 使用assert断言
	assert A == B

测试

cd src
pytest test

# 需要显示print
pytest -s test

进度条

x = 10000
for i in range(x):
    for j in range(x):
        k = j * i
    print(f"{i + 1} / {x}", end="\r")
print("\n完成!")

end="\r"表示每次输出后都不换行

第三方库:

• tqdm
• progress

python vscode调试

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: Main",
            "type": "python",
            "request": "launch",
            "program": "${workspaceFolder}/main.py",  // main.py 的路径
            "console": "integratedTerminal",
            "justMyCode": false,  // 设置为 false 以允许在子模块中设置断点
            "env": {
                // 在这里添加环境变量（如果需要）
            }
        }
    ]
}