AI 教我学 Python（03）：Python 内置库与生态、单元测试、开发与部署规范

name: str = "Alice"
age: int = 30
height: float = 5.8
is_active: bool = True
ids: list = [1, 2, 3] # 基本，但不够具体
config: dict = {"key": "value"} # 基本，不够具体
coords: tuple = (10, 20)
unique_ids: set = {1, 2, 3}
data: bytes = b"hello"
maybe_value: None = None # 明确是 None

from typing import Any

flexible_data: Any = "could be anything"
flexible_data = 123
flexible_data = [1, "a"]

def process_anything(data: Any) -> Any:
    # 类型检查器不会抱怨这里的操作
    print(data)
    return data

from typing import Callable

# 一个不接受参数且返回 None 的函数
simple_callback: Callable[[], None] = lambda: print("Called!")

# 一个接受两个 int 并返回 float 的函数
calculator: Callable[[int, int], float] = lambda x, y: float(x + y)

# 一个接受一个字符串和可选整数，返回布尔值的函数
validator: Callable[[str, int | None], bool] = lambda s, n: (n is None or len(s) > n)

def execute_callback(cb: Callable[[str], None], value: str) -> None:
    print("Executing callback...")
    cb(value)
    print("Callback finished.")

execute_callback(lambda s: print(f"Processed: {s}"), "hello")

from typing import TypeVar, List

T = TypeVar('T') # 声明一个类型变量 'T'

def get_first(items: List[T]) -> T: # 函数接受 T 类型的列表，返回 T 类型的值
    return items[0]

first_str: str = get_first(["a", "b", "c"]) # T 被推断为 str
first_int: int = get_first([1, 2, 3])       # T 被推断为 int

# 也可以约束 TypeVar
NumberT = TypeVar('NumberT', int, float, complex) # T 只能是 int, float, 或 complex

def add(a: NumberT, b: NumberT) -> NumberT:
    return a + b

sum_int: int = add(1, 2)
sum_float: float = add(1.5, 2.5)
# sum_str = add("a", "b") # 这会被类型检查器标记为错误

from typing import NewType

UserId = NewType('UserId', int)
OrderId = NewType('OrderId', int)

def process_user(user_id: UserId) -> None:
    print(f"Processing user with ID: {user_id}")

def process_order(order_id: OrderId) -> None:
    print(f"Processing order with ID: {order_id}")

user_a = UserId(123)
order_b = OrderId(456)

process_user(user_a)
process_order(order_b)

# 类型检查器会报错，因为类型不匹配，即使它们都是 int
# process_user(order_b)
# process_order(user_a)

# 在运行时，它们仍然是 int
print(type(user_a)) # <class 'int'>
assert user_a == 123

from typing import Protocol, List

class Serializable(Protocol):
    def serialize(self) -> str:
        ... # 省略号表示这是一个协议方法，不需要实现

class User:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

    def serialize(self) -> str: # 实现了 serialize 方法
        return f'{{"name": "{self.name}", "age": {self.age}}}'

class Product:
    def __init__(self, name: str, price: float):
        self.name = name
        self.price = price
    # 没有 serialize 方法

def save_serialized(items: List[Serializable]) -> None:
    for item in items:
        data = item.serialize() # 类型检查器知道 item 有 serialize 方法
        print(f"Saving: {data}")

user1 = User("Alice", 30)
user2 = User("Bob", 25)
# product1 = Product("Laptop", 1200.0) # 类型检查器会报错，因为它不符合 Serializable 协议

save_serialized([user1, user2])
# save_serialized([product1]) # 这会引发类型错误

from typing import Literal

Mode = Literal["read", "write", "append"]

def open_file(file_path: str, mode: Mode) -> None:
    print(f"Opening {file_path} in mode '{mode}'")

open_file("data.txt", "read")
open_file("log.txt", "append")
# open_file("config.ini", "execute") # 类型检查器会报错，因为 "execute" 不是允许的 Mode 值

from typing import TypedDict

class Point(TypedDict):
    x: int
    y: int
    label: str | None # 可以包含可选键

point: Point = {'x': 10, 'y': 20, 'label': 'A'}
point2: Point = {'x': 5, 'y': 15} # label 是可选的

def process_point(p: Point) -> None:
    print(f"Processing point at ({p['x']}, {p['y']}) with label '{p.get('label')}'")

process_point(point)
process_point(point2)

# 类型检查器会报错：
# wrong_point: Point = {'x': 1, 'z': 2} # 键名错误
# wrong_type: Point = {'x': '10', 'y': 20} # 'x' 类型错误

from typing import Final

API_KEY: Final[str] = "your_secret_api_key"
DEFAULT_TIMEOUT: Final = 10 # 类型可以被推断

# API_KEY = "new_key" # 类型检查器会报错

class Config:
    VERSION: Final[str] = "1.0"

# Config.VERSION = "1.1" # 类型检查器会报错

squares = [x**2 for x in range(10)]  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
evens = [x for x in range(10) if x % 2 == 0]  # [0, 2, 4, 6, 8]

gen = (x**2 for x in range(10))  # 创建一个生成器对象，不立即计算

// JS 等价实现
Array.from(
  { length: length },
  () => characters[Math.floor(Math.random() * characters.length)]
).join('');

lst = [0, 1, 2, 3, 4, 5]
print(lst[2:5])    # [2, 3, 4]
print(lst[::-1])   # [5, 4, 3, 2, 1, 0] (反转)

a, b = 1, 2
a, b = b, a  # 交换值，无需临时变量
first, *rest, last = [1, 2, 3, 4, 5]  # first=1, rest=[2,3,4], last=5

@cache
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

with open('file.txt', 'r') as f:
    content = f.read()
# 文件自动关闭

def function(pos_arg, *args, kwarg='default', **kwargs):
    # pos_arg：位置参数
    # *args：可变位置参数
    # kwarg：关键字参数，有默认值
    # **kwargs：可变关键字参数
    pass

def greeting(name: str) -> str:
    return f"Hello, {name}"

def append_to(element, lst=[]):  # 危险！
    lst.append(element)
    return lst

print(append_to(1))  # [1]
print(append_to(2))  # [1, 2] 而非预期的 [2]

x = 10
def outer():
    x = 20  # 创建局部变量，不影响全局变量
    def inner():
        print(x)  # 在Python 3中可以访问外部的x，但不能修改
    inner()

a = [1, 2, 3]
b = [1, 2, 3]
c = a

print(a == b)  # True（值相等）
print(a is b)  # False（不是同一对象）
print(a is c)  # True（是同一对象）

a = 256
b = 256
print(a is b)  # 可能是True或False，取决于实现

c = 5
d = 5
print(c is d)  # 通常是True，因为小整数缓存

my_project/
│
├── pyproject.toml      # 现代Python项目配置（类似package.json）
├── setup.py            # 遗留的包安装脚本（类似老式npm项目）
├── requirements.txt    # 依赖列表（类似package.json的dependencies）
├── README.md           # 项目说明
├── LICENSE             # 许可证
├── .gitignore
├── .pre-commit-config.yaml  # 预提交钩子配置
│
├── my_package/         # 主包目录（实际代码）
│   ├── __init__.py     # 声明此目录为Python包
│   ├── module1.py
│   ├── module2.py
│   └── subpackage/
│       ├── __init__.py
│       └── module3.py
│
├── tests/              # 测试目录
│   ├── __init__.py
│   ├── test_module1.py
│   └── test_module2.py
│
└── docs/               # 文档目录
    ├── conf.py
    ├── index.rst
    └── ...

python -m venv .venv
source .venv/bin/activate  # Linux/Mac
.venv\Scripts\activate     # Windows

# requirements.txt - 基础依赖
flask==2.2.3

# requirements-dev.txt - 开发依赖
-r requirements.txt  # 包含基础依赖
pytest==7.3.1
black==23.3.0

def fetch_data(url, timeout=30):
    """获取指定URL的数据。

    获取URL指向的数据并返回解析后的JSON对象。
    如果请求失败，将抛出异常。

    Args:
        url: 需要请求的URL地址。
        timeout: 请求超时时间，默认30秒。

    Returns:
        dict: 解析后的JSON数据。

    Raises:
        RequestError: 如果请求失败。
    """

pip install sphinx sphinx-rtd-theme
sphinx-quickstart docs

# .github/workflows/python-app.yml
name: Python application

on: [push, pull_request]

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install flake8 pytest
        pip install -r requirements.txt
    - name: Lint with flake8
      run: flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
    - name: Test with pytest
      run: pytest

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def main():
    logger.info("应用启动")
    try:
        # 业务逻辑
        result = complex_operation()
        logger.info(f"操作完成，结果：{result}")
    except Exception as e:
        logger.exception("发生错误")

# .pre-commit-config.yaml
repos:
-   repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.4.0
    hooks:
    -   id: trailing-whitespace
    -   id: end-of-file-fixer
    -   id: check-yaml
    -   id: check-added-large-files

-   repo: https://github.com/psf/black
    rev: 23.3.0
    hooks:
    -   id: black

-   repo: https://github.com/PyCQA/flake8
    rev: 6.0.0
    hooks:
    -   id: flake8

-   repo: https://github.com/PyCQA/isort
    rev: 5.12.0
    hooks:
    -   id: isort

[build-system]
requires = ["setuptools>=42", "wheel"]
build-backend = "setuptools.build_meta"

[tool.black]
line-length = 88
target-version = ["py38", "py39", "py310"]

[tool.isort]
profile = "black"
multi-line-output = 3

[tool.mypy]
python-version = "3.8"
warn-return-any = true
warn-unused-configs = true
disallow-untyped-defs = true
disallow-incomplete-defs = true

[tool.pytest.ini-options]
testpaths = ["tests"]
python-files = "test_*.py"
addopts = "--cov=my_package"

# 简单的Django视图
from django.http import HttpResponse

def hello_world(request):
    return HttpResponse("Hello, World!")

# Flask快速入门
from flask import Flask
app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello, World!'

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

# FastAPI示例
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

import numpy as np

# 创建数组
arr = np.array([1, 2, 3, 4, 5])

# 基本操作
print(arr * 2)  # [2 4 6 8 10]
print(arr.mean())  # 3.0

import pandas as pd

# 创建DataFrame
df = pd.DataFrame({
    'A': [1, 2, 3],
    'B': ['a', 'b', 'c']
})

# 数据操作
filtered = df[df['A'] > 1]

import matplotlib.pyplot as plt

# 简单线图
plt.plot([1, 2, 3, 4], [1, 4, 9, 16])
plt.xlabel('X轴')
plt.ylabel('Y轴')
plt.title('示例图表')
plt.show()

import requests

# 发送GET请求
response = requests.get('https://api.github.com')

# 解析JSON响应
data = response.json()

# POST请求
response = requests.post(
    'https://httpbin.org/post',
    json={'key': 'value'}
)

from bs4 import BeautifulSoup

# 解析HTML
soup = BeautifulSoup('<html><body><p>Hello World</p></body></html>', 'html.parser')

# 提取数据
print(soup.p.text)  # 输出: Hello World

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 建立连接
engine = create_engine('sqlite:///example.db')
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

# 创建表
Base.metadata.create_all(engine)

# 添加数据
Session = sessionmaker(bind=engine)
session = Session()
new_user = User(name='Alice', age=30)
session.add(new_user)
session.commit()

from pymongo import MongoClient

# 连接MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['customers']

# 插入文档
customer = {"name": "John", "age": 30}
result = collection.insert_one(customer)

# 查询文档
for doc in collection.find({"age": {"$gt": 25}}):
    print(doc)

import subprocess

# 执行系统命令
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print(result.stdout)

from selenium import webdriver

# 启动浏览器
driver = webdriver.Chrome()

# 打开网页
driver.get('https://www.example.com')

# 查找元素并交互
element = driver.find_element_by_id('search')
element.send_keys('test')
element.submit()

# 关闭浏览器
driver.quit()

from PIL import Image, ImageFilter

# 打开图像
img = Image.open('example.jpg')

# 应用滤镜
blurred = img.filter(ImageFilter.BLUR)

# 调整大小
resized = img.resize((300, 300))

# 保存
resized.save('resized_image.jpg')

import openpyxl

# 创建工作簿
wb = openpyxl.Workbook()
sheet = wb.active

# 写入数据
sheet['A1'] = '姓名'
sheet['B1'] = '年龄'
sheet['A2'] = '张三'
sheet['B2'] = 25

# 保存
wb.save('example.xlsx')

# 读取
wb = openpyxl.load_workbook('example.xlsx')
sheet = wb.active
print(sheet['A2'].value)  # 输出: 张三

from pydantic import BaseModel, EmailStr, ValidationError

class User(BaseModel):
    id: int
    name: str
    email: EmailStr
    age: int = None

# 验证有效数据
try:
    user = User(id=1, name="John", email="john@example.com")
    print(user.dict())
except ValidationError as e:
    print(e.json())

# 验证无效数据
try:
    User(id="not_an_int", name="John", email="not_an_email")
except ValidationError as e:
    print(e.json())

# .env文件
# API_KEY=your_secret_key
# DEBUG=True

from dotenv import load_dotenv
import os

# 加载.env文件
load_dotenv()

# 使用环境变量
api_key = os.getenv("API_KEY")
debug = os.getenv("DEBUG") == "True"

import asyncio
import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://example.com", "https://python.org", "https://github.com"]
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for url, result in zip(urls, results):
        print(f"{url}: 内容长度 {len(result)} 字符")

# 运行异步函数
asyncio.run(main())

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# 加载数据集
iris = load_iris()
X, y = iris.data, iris.target

# 分割训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3)

# 训练模型
model = RandomForestClassifier()
model.fit(X_train, y_train)

# 预测和评估
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
print(f"准确率: {accuracy:.2f}")

# PyTorch简单示例
import torch
import torch.nn as nn

# 定义一个简单的神经网络
class SimpleNN(nn.Module):
    def __init__(self):
        super(SimpleNN, self).__init__()
        self.fc1 = nn.Linear(10, 5)
        self.fc2 = nn.Linear(5, 1)
        self.relu = nn.ReLU()

    def forward(self, x):
        x = self.relu(self.fc1(x))
        x = self.fc2(x)
        return x

# 创建模型实例
model = SimpleNN()
print(model)

# test_simple.py
def test_addition():
    assert 1 + 1 == 2

def test_string():
    assert "hello".upper() == "HELLO"

pytest
# 或运行特定文件
pytest test_simple.py

// Jest示例
test('addition works', () => {
  expect(1 + 1).toBe(2);
});

test('string uppercasing works', () => {
  expect('hello'.toUpperCase()).toBe('HELLO');
});