Python 开发指南

环境准备

Python 版本要求

推荐使用 Python 3.8+ 以获得最佳的开发体验和语言特性支持。

# 检查当前 Python 版本
python --version
python3 --version

# 推荐安装最新稳定版本
# 从 https://www.python.org/downloads/ 下载安装

必备工具安装

核心工具

# pip - Python 包管理器（通常随 Python 安装）
python -m pip --version

# 升级 pip 到最新版本
python -m pip install --upgrade pip

# 虚拟环境工具（推荐）
python -m pip install virtualenv

# 或者使用内置的 venv
python -m venv --help

高级包管理器

# Poetry - 现代依赖管理和打包工具
curl -sSL https://install.python-poetry.org | python3 -

# Conda - 科学计算和数据科学环境管理
# 从 https://docs.conda.io/en/latest/miniconda.html 下载安装

# pipenv - 简化的虚拟环境管理
pip install pipenv

项目配置和结构

虚拟环境管理

使用 venv（推荐初学者）

# 创建虚拟环境
python -m venv myproject-env

# 激活虚拟环境
# Windows
myproject-env\Scripts\activate
# macOS/Linux
source myproject-env/bin/activate

# 验证环境
which python
pip list

# 停用环境
deactivate

使用 Poetry（推荐专业开发）

# 初始化新项目
poetry new myproject
cd myproject

# 在现有项目中初始化
cd existing-project
poetry init

# 安装依赖
poetry install

# 激活虚拟环境
poetry shell

# 添加依赖
poetry add requests pandas
poetry add --group dev pytest black flake8

使用 Conda（推荐数据科学）

# 创建新环境
conda create -n myproject python=3.11

# 激活环境
conda activate myproject

# 安装包
conda install pandas numpy matplotlib jupyter
conda install -c conda-forge fastapi

# 导出环境
conda env export > environment.yml

# 从文件创建环境
conda env create -f environment.yml

项目结构最佳实践

标准 Python 项目结构

myproject/
├── src/
│   └── myproject/
│       ├── __init__.py
│       ├── main.py
│       ├── models/
│       │   ├── __init__.py
│       │   └── user.py
│       ├── services/
│       │   ├── __init__.py
│       │   └── user_service.py
│       └── utils/
│           ├── __init__.py
│           └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── test_models/
│   └── test_services/
├── docs/
├── scripts/
├── requirements.txt
├── pyproject.toml
├── README.md
└── .gitignore

Web 应用项目结构（Flask/FastAPI）

webapp/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── routes/
│   │   └── middleware/
│   ├── models/
│   ├── services/
│   ├── schemas/
│   └── database/
├── tests/
├── migrations/
├── static/
├── templates/
├── config/
├── requirements.txt
└── Dockerfile

数据科学项目结构

data-project/
├── data/
│   ├── raw/
│   ├── processed/
│   └── external/
├── notebooks/
│   ├── exploratory/
│   ├── modeling/
│   └── visualization/
├── src/
│   ├── data/
│   ├── features/
│   ├── models/
│   └── visualization/
├── models/
├── reports/
├── requirements.txt
└── environment.yml

Kiro 核心功能应用

项目初始化

使用 Kiro 快速初始化 Python 项目：

# 在 Kiro 聊天中请求：
"""
请帮我创建一个 FastAPI 项目结构，包含以下功能：
- 用户认证
- RESTful API 端点
- 数据库集成（PostgreSQL）
- Docker 配置
- 测试设置
- 代码质量工具配置
"""

代码分析和改进

PEP 8 合规性检查

# Kiro 会检测并建议修复的代码问题

# 原始代码（不符合 PEP 8）
def calculateUserAge(userBirthDate,currentDate=None):
    if currentDate==None:
        currentDate=datetime.now()
    age=(currentDate-userBirthDate).days//365
    return age

# Kiro 建议的改进版本
from datetime import datetime
from typing import Optional

def calculate_user_age(
    user_birth_date: datetime, 
    current_date: Optional[datetime] = None
) -> int:
    """
    计算用户年龄。
    
    Args:
        user_birth_date: 用户出生日期
        current_date: 当前日期，默认为现在
        
    Returns:
        用户年龄（年）
    """
    if current_date is None:
        current_date = datetime.now()
    
    age = (current_date - user_birth_date).days // 365
    return age

性能优化建议

# Kiro 可以识别性能问题并提供优化方案

# 问题代码：低效的列表操作
def process_large_dataset(data):
    result = []
    for item in data:
        if item['status'] == 'active':
            processed = expensive_operation(item)
            result.append(processed)
    return result

# Kiro 建议的优化版本
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor

def process_large_dataset(data: List[Dict[str, Any]]) -> List[Any]:
    """使用生成器和并行处理优化大数据集处理。"""
    # 使用生成器表达式过滤
    active_items = (item for item in data if item['status'] == 'active')
    
    # 并行处理
    with ThreadPoolExecutor(max_workers=4) as executor:
        result = list(executor.map(expensive_operation, active_items))
    
    return result

# 或者使用内置函数优化
def process_large_dataset_v2(data: List[Dict[str, Any]]) -> List[Any]:
    """使用内置函数优化处理。"""
    return [
        expensive_operation(item) 
        for item in data 
        if item['status'] == 'active'
    ]

类型提示建议

# Kiro 会建议添加类型提示

# 原始代码
def fetch_user_data(user_id, include_posts=True):
    user = db.get_user(user_id)
    if include_posts:
        user['posts'] = db.get_user_posts(user_id)
    return user

# Kiro 建议的类型注解版本
from typing import Dict, Any, Optional, List

UserData = Dict[str, Any]
PostData = Dict[str, Any]

def fetch_user_data(
    user_id: int, 
    include_posts: bool = True
) -> Optional[UserData]:
    """
    获取用户数据。
    
    Args:
        user_id: 用户 ID
        include_posts: 是否包含用户帖子
        
    Returns:
        用户数据字典，如果用户不存在则返回 None
    """
    user: Optional[UserData] = db.get_user(user_id)
    if user is None:
        return None
        
    if include_posts:
        posts: List[PostData] = db.get_user_posts(user_id)
        user['posts'] = posts
        
    return user

调试和错误处理

异常处理改进

# Kiro 建议的错误处理模式

# 原始代码
def divide_numbers(a, b):
    return a / b

# Kiro 建议的改进版本
from typing import Union
import logging

logger = logging.getLogger(__name__)

def divide_numbers(a: float, b: float) -> Union[float, None]:
    """
    安全的数字除法操作。
    
    Args:
        a: 被除数
        b: 除数
        
    Returns:
        除法结果，如果出错则返回 None
    """
    try:
        if b == 0:
            raise ValueError("除数不能为零")
        return a / b
    except (TypeError, ValueError) as e:
        logger.error(f"除法操作错误: {e}")
        return None
    except Exception as e:
        logger.error(f"未知错误: {e}")
        raise

代理钩子配置

自动测试生成钩子

name: "Python 测试生成器"
trigger: onSave
filePattern: "src/**/*.py"
instructions: |
  为新的或修改的 Python 函数生成 pytest 测试用例：
  
  1. 分析函数的参数、返回值和异常
  2. 生成基本功能测试
  3. 添加边界条件和错误情况测试
  4. 包含模拟对象测试（如果有外部依赖）
  5. 确保测试覆盖主要代码路径
  
  测试应该遵循 AAA 模式（Arrange, Act, Assert）

依赖更新检查钩子

name: "Python 依赖检查"
trigger: manual
instructions: |
  检查 Python 项目的依赖更新和安全问题：
  
  1. 运行 `pip list --outdated` 检查过时包
  2. 使用 `safety check` 扫描安全漏洞
  3. 检查 requirements.txt 或 pyproject.toml
  4. 提供更新建议和风险评估
  5. 生成依赖更新计划

代码质量检查钩子

name: "Python 代码质量检查"
trigger: onSave
filePattern: "**/*.py"
instructions: |
  执行全面的 Python 代码质量检查：
  
  1. 运行 flake8 或 pylint 进行代码检查
  2. 使用 black 检查代码格式
  3. 运行 isort 检查导入排序
  4. 使用 mypy 进行类型检查
  5. 检查文档字符串的完整性
  6. 提供具体的改进建议

虚拟环境管理钩子

name: "虚拟环境管理器"
trigger: onSave
filePattern: "requirements.txt"
instructions: |
  管理 Python 虚拟环境和依赖：
  
  1. 检测 requirements.txt 变更
  2. 提醒激活正确的虚拟环境
  3. 建议运行 pip install -r requirements.txt
  4. 检查版本冲突
  5. 提供环境同步建议

框架特定配置

Django 项目配置

settings.py 最佳实践

# settings/base.py
import os
from pathlib import Path
from decouple import config
from dj_database_url import parse as db_url

BASE_DIR = Path(__file__).resolve().parent.parent.parent

# 安全配置
SECRET_KEY = config('SECRET_KEY')
DEBUG = config('DEBUG', default=False, cast=bool)
ALLOWED_HOSTS = config('ALLOWED_HOSTS', cast=lambda v: [s.strip() for s in v.split(',')])

# 应用配置
DJANGO_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
]

THIRD_PARTY_APPS = [
    'rest_framework',
    'corsheaders',
    'django_filters',
]

LOCAL_APPS = [
    'apps.users',
    'apps.core',
]

INSTALLED_APPS = DJANGO_APPS + THIRD_PARTY_APPS + LOCAL_APPS

# 数据库配置
DATABASES = {
    'default': config(
        'DATABASE_URL',
        cast=db_url,
        default='sqlite:///' + str(BASE_DIR / 'db.sqlite3')
    )
}

# REST Framework 配置
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.TokenAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 20,
}

FastAPI 项目配置

main.py 结构

from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
import uvicorn

from app.database import get_db
from app.routers import users, items
from app.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    description=settings.PROJECT_DESCRIPTION,
    version=settings.VERSION,
    openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

# 中间件配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_HOSTS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 路由注册
app.include_router(
    users.router,
    prefix=f"{settings.API_V1_STR}/users",
    tags=["users"]
)

app.include_router(
    items.router,
    prefix=f"{settings.API_V1_STR}/items",
    tags=["items"]
)

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

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True
    )

Pydantic 模型

from pydantic import BaseModel, EmailStr, validator
from typing import Optional, List
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    username: str
    full_name: Optional[str] = None
    is_active: bool = True

class UserCreate(UserBase):
    password: str
    
    @validator('password')
    def validate_password(cls, v):
        if len(v) < 8:
            raise ValueError('密码长度至少为8位')
        return v

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    username: Optional[str] = None
    full_name: Optional[str] = None
    is_active: Optional[bool] = None

class UserInDB(UserBase):
    id: int
    created_at: datetime
    updated_at: datetime
    
    class Config:
        orm_mode = True

class UserResponse(UserInDB):
    pass

数据科学和机器学习

Jupyter Notebook 最佳实践

Notebook 组织结构

# 1. 导入和配置
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier

# 配置显示选项
pd.set_option('display.max_columns', None)
plt.style.use('seaborn-v0_8')
sns.set_palette("husl")

# 2. 数据加载和初步探索
def load_and_explore_data(file_path: str) -> pd.DataFrame:
    """加载数据并进行初步探索。"""
    df = pd.read_csv(file_path)
    
    print(f"数据形状: {df.shape}")
    print(f"缺失值:\n{df.isnull().sum()}")
    print(f"数据类型:\n{df.dtypes}")
    
    return df

# 3. 数据清洗和预处理
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """数据清洗和预处理。"""
    # 处理缺失值
    df = df.dropna(subset=['target_column'])
    
    # 数据类型转换
    df['date_column'] = pd.to_datetime(df['date_column'])
    
    # 特征工程
    df['feature_engineered'] = df['feature1'] * df['feature2']
    
    return df

# 4. 可视化分析
def create_visualizations(df: pd.DataFrame) -> None:
    """创建数据可视化。"""
    fig, axes = plt.subplots(2, 2, figsize=(15, 10))
    
    # 分布图
    df['target'].hist(ax=axes[0, 0], bins=30)
    axes[0, 0].set_title('目标变量分布')
    
    # 相关性热图
    correlation_matrix = df.corr()
    sns.heatmap(correlation_matrix, ax=axes[0, 1], annot=True)
    
    # 特征重要性
    # ... 其他可视化
    
    plt.tight_layout()
    plt.show()

机器学习工作流

模型训练和评估

from sklearn.metrics import classification_report, confusion_matrix
from sklearn.model_selection import cross_val_score
import joblib

class ModelTrainer:
    """机器学习模型训练器。"""
    
    def __init__(self, model=None):
        self.model = model or RandomForestClassifier(n_estimators=100, random_state=42)
        self.is_trained = False
    
    def train(self, X_train: np.ndarray, y_train: np.ndarray) -> None:
        """训练模型。"""
        print("开始训练模型...")
        self.model.fit(X_train, y_train)
        self.is_trained = True
        print("模型训练完成")
    
    def evaluate(self, X_test: np.ndarray, y_test: np.ndarray) -> dict:
        """评估模型性能。"""
        if not self.is_trained:
            raise ValueError("模型尚未训练")
        
        predictions = self.model.predict(X_test)
        
        # 交叉验证分数
        cv_scores = cross_val_score(self.model, X_test, y_test, cv=5)
        
        # 分类报告
        class_report = classification_report(y_test, predictions, output_dict=True)
        
        # 混淆矩阵
        conf_matrix = confusion_matrix(y_test, predictions)
        
        return {
            'cv_scores': cv_scores,
            'cv_mean': cv_scores.mean(),
            'cv_std': cv_scores.std(),
            'classification_report': class_report,
            'confusion_matrix': conf_matrix
        }
    
    def save_model(self, file_path: str) -> None:
        """保存训练好的模型。"""
        if not self.is_trained:
            raise ValueError("模型尚未训练")
        
        joblib.dump(self.model, file_path)
        print(f"模型已保存到: {file_path}")
    
    @classmethod
    def load_model(cls, file_path: str):
        """加载已保存的模型。"""
        model = joblib.load(file_path)
        trainer = cls(model)
        trainer.is_trained = True
        return trainer

测试策略

pytest 配置

conftest.py

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.database import get_db, Base

# 测试数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture
def db_session():
    """创建测试数据库会话。"""
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)

@pytest.fixture
def client(db_session):
    """创建测试客户端。"""
    def override_get_db():
        try:
            yield db_session
        finally:
            db_session.close()
    
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client

@pytest.fixture
def sample_user_data():
    """示例用户数据。"""
    return {
        "email": "test@example.com",
        "username": "testuser",
        "password": "testpassword123",
        "full_name": "Test User"
    }

单元测试示例

import pytest
from app.services.user_service import UserService
from app.models.user import User

class TestUserService:
    """用户服务测试类。"""
    
    def test_create_user_success(self, db_session, sample_user_data):
        """测试成功创建用户。"""
        # Arrange
        user_service = UserService(db_session)
        
        # Act
        user = user_service.create_user(sample_user_data)
        
        # Assert
        assert user.email == sample_user_data["email"]
        assert user.username == sample_user_data["username"]
        assert user.id is not None
    
    def test_create_user_duplicate_email(self, db_session, sample_user_data):
        """测试创建重复邮箱用户失败。"""
        # Arrange
        user_service = UserService(db_session)
        user_service.create_user(sample_user_data)
        
        # Act & Assert
        with pytest.raises(ValueError, match="邮箱已存在"):
            user_service.create_user(sample_user_data)
    
    @pytest.mark.parametrize("invalid_email", [
        "invalid-email",
        "@example.com",
        "test@",
        ""
    ])
    def test_create_user_invalid_email(self, db_session, sample_user_data, invalid_email):
        """测试无效邮箱创建用户失败。"""
        # Arrange
        user_service = UserService(db_session)
        sample_user_data["email"] = invalid_email
        
        # Act & Assert
        with pytest.raises(ValueError, match="无效的邮箱格式"):
            user_service.create_user(sample_user_data)

部署和生产配置

Docker 配置

Dockerfile

FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建非root用户
RUN useradd --create-home --shell /bin/bash app \
    && chown -R app:app /app
USER app

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yml

version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/myapp
    depends_on:
      - db
    volumes:
      - ./app:/app
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

  db:
    image: postgres:15
    environment:
      POSTGRES_DB: myapp
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:

生产环境配置

环境变量管理

# config/settings.py
from pydantic import BaseSettings
from typing import List

class Settings(BaseSettings):
    """应用配置。"""
    
    # 基础配置
    PROJECT_NAME: str = "My Python App"
    VERSION: str = "1.0.0"
    DEBUG: bool = False
    
    # 数据库配置
    DATABASE_URL: str
    
    # 安全配置
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    # 外部服务
    REDIS_URL: str = "redis://localhost:6379"
    
    # CORS 配置
    ALLOWED_HOSTS: List[str] = ["localhost", "127.0.0.1"]
    
    # 日志配置
    LOG_LEVEL: str = "INFO"
    
    class Config:
        env_file = ".env"
        case_sensitive = True

settings = Settings()