摘要："""完整的Python项目结构示例项目名称: TaskMaster - 任务管理系统"""import osimport sysfrom pathlib import Pathfrom typing import Dict, List, Optionalfr
今天我们将学习Python项目开发的工程化实践，包括项目结构、配置管理、打包发布、文档编写和团队协作。
"""完整的Python项目结构示例项目名称: TaskMaster - 任务管理系统"""import osimport sysfrom pathlib import Pathfrom typing import Dict, List, Optionalfrom dataclasses import dataclass, asdictimport jsonimport yamlfrom datetime import datetimeimport loggingfrom logging.config import dictConfig# 项目基础结构def create_project_structure: """创建标准的Python项目结构""" print("=== Python项目结构设计 ===") project_structure = { 'taskmaster/': { '__init__.py': '# 主包初始化', 'src/': { '__init__.py': '', 'core/': { '__init__.py': '# 核心功能模块', 'models.py': '# 数据模型', 'services.py': '# 业务逻辑服务', 'exceptions.py': '# 自定义异常', }, 'api/': { '__init__.py': '# API模块', 'routes.py': '# 路由定义', 'schemas.py': '# API数据模式', 'dependencies.py': '# 依赖注入', }, 'database/': { '__init__.py': '# 数据库模块', 'connection.py': '# 数据库连接', 'repositories.py': '# 数据访问层', 'migrations/': { '__init__.py': '# 数据库迁移', }, }, 'utils/': { '__init__.py': '# 工具函数', 'validators.py': '# 数据验证', 'helpers.py': '# 辅助函数', 'security.py': '# 安全相关', }, }, 'tests/': { '__init__.py': '# 测试包', 'conftest.py': '# pytest配置', 'test_models.py': '# 模型测试', 'test_services.py': '# 服务测试', 'test_api/': { '__init__.py': '', 'test_routes.py': '# API路由测试', }, 'integration/': { '__init__.py': '# 集成测试', 'test_DATABASE.py': '# 数据库集成测试', }, }, 'docs/': { 'conf.py': '# Sphinx配置', 'index.rst': '# 文档首页', 'api.rst': '# API文档', 'tutorial.rst': '# 使用教程', }, 'scripts/': { 'setup_database.py': '# 数据库初始化脚本', 'deploy.py': '# 部署脚本', 'backup.py': '# 备份脚本', }, 'config/': { '__init__.py': '', 'default.py': '# 默认配置', 'development.py': '# 开发环境配置', 'production.py': '# 生产环境配置', 'testing.py': '# 测试环境配置', }, 'alembic/': { 'versions/': {'__init__.py': ''}, 'env.py': '# Alembic环境配置', 'script.py.mako': '# 迁移模板', }, }, 'requirements/': { 'base.txt': '# 基础依赖', 'dev.txt': '# 开发依赖', 'prod.txt': '# 生产依赖', 'test.txt': '# 测试依赖', }, '.github/': { 'workflows/': { 'ci.yml': '# CI流水线', 'cd.yml': '# CD流水线', }, }, } # 项目配置文件 config_files = { 'pyproject.toml': '''[build-system]requires = ["setuptools>=45", "wheel"]build-backend = "setuptools.build_meta"[project]name = "taskmaster"version = "0.1.0"description = "A modern task management system"authors = [ {name = "Your Name", email = "your.email@example.com"}]readme = "README.md"license = {text = "MIT"}requires-python = ">=3.8"dependencies = [ "fastapi>=0.68.0", "sqlalchemy>=1.4.0", "pydantic>=1.8.0", "alembic>=1.7.0",][project.optional-dependencies]dev = [ "pytest>=6.0", "black>=21.0", "flake8>=3.9", "mypy>=0.910",]test = [ "pytest-cov>=2.0", "httpx>=0.19.0",][tool.black]line-length = 88target-version = ['py38'][tool.mypy]python_version = "3.8"warn_return_any = truewarn_unused_configs = true[tool.pytest.ini_options]testpaths = ["tests"]addopts = "-v --cov=taskmaster --cov-report=html"''', 'setup.py': '''from setuptools import setup, find_packagessetup( name="taskmaster", version="0.1.0", packages=find_packages, install_requires=[ "fastapi>=0.68.0", "sqlalchemy>=1.4.0", "pydantic>=1.8.0", ], extras_require={ "dev": [ "pytest>=6.0", "black>=21.0", "flake8>=3.9", ], "test": [ "pytest-cov>=2.0", ], }, python_requires=">=3.8",)''', 'Dockerfile': '''FROM python:3.9-slimWORKDIR /app# 安装系统依赖RUN apt-get update && apt-get install -y \ gcc \ && rm -rf /var/lib/apt/lists/*# 复制依赖文件COPY requirements/prod.txt .# 安装Python依赖RUN pip install --no-cache-dir -r prod.txt# 复制应用代码COPY taskmaster/ .# 创建非root用户RUN useradd --create-home --shell /bin/bash appUSER app# 暴露端口EXPOSE 8000# 启动应用CMD ["uvicorn", "taskmaster.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/taskmaster depends_on: - db volumes: - ./taskmaster:/app db: image: postgres:13 environment: - POSTGRES_USER=user - POSTGRES_PASSWORD=password - POSTGRES_DB=taskmaster volumes: - postgres_data:/var/lib/postgresql/datavolumes: postgres_data:''', 'README.md': '''# TaskMaster现代化的任务管理系统## 特性- 快速的任务创建和管理- 安全的用户认证- 丰富的统计报表- 实时数据同步## 安装```bashpip install -e .# 安装开发依赖pip install -e ".[dev]"# 运行测试pytest# 代码格式化black taskmaster/使用Docker部署
docker-compose up -d''',
'.gitignore': '''
pycache/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
.vscode/
.idea/
*.swp
*.swo
*.db
*.sqlite3
*.log
logs/
''',
'.env.example': '''
DATABASE_URL=postgresql://user:password@localhost/taskmaster
第三方服务SENTRY_DSN=your-sentry-dsn
EMAIL_HOST=smtp.gmail.com
EMAIL_PORT=587
''',
}
project_structure, config_files = create_project_structure
### 练习2：配置管理与环境设置```python"""配置管理和环境设置"""import osfrom enum import Enumfrom typing import Any, Dictfrom pydantic import BaseSettings, validatorclass Environment(str, Enum): """环境类型枚举""" DEVELOPMENT = "development" TESTING = "testing" PRODUCTION = "production"class DatabaseConfig: """数据库配置类""" def __init__(self, url: str, echo: bool = False, pool_size: int = 5): self.url = url self.echo = echo self.pool_size = pool_size @classmethod def from_env(cls) -> 'DatabaseConfig': """从环境变量创建配置""" url = os.getenv('DATABASE_URL', 'sqlite:///./taskmaster.db') echo = os.getenv('DATABASE_ECHO', 'false').lower == 'true' pool_size = int(os.getenv('DATABASE_POOL_SIZE', '5')) return cls(url=url, echo=echo, pool_size=pool_size)class SecurityConfig: """安全配置""" def __init__(self, secret_key: str, algorithm: str = "HS256", access_token_expire_minutes: int = 30): self.secret_key = secret_key self.algorithm = algorithm self.access_token_expire_minutes = access_token_expire_minutes @classmethod def from_env(cls) -> 'SecurityConfig': """从环境变量创建安全配置""" secret_key = os.getenv('SECRET_KEY') if not secret_key: raise ValueError("SECRET_KEY环境变量必须设置") algorithm = os.getenv('JWT_algorithm', 'HS256') expire_minutes = int(os.getenv('ACCESS_TOKEN_EXPIRE_MINUTES', '30')) return cls( secret_key=secret_key, algorithm=algorithm, access_token_expire_minutes=expire_minutes )class Settings(BaseSettings): """应用设置 - 使用pydantic进行验证""" # 基础配置 app_name: str = "TaskMaster" environment: Environment = Environment.DEVELOPMENT debug: bool = False # 数据库配置 database_url: str = "sqlite:///./taskmaster.db" database_echo: bool = False # 安全配置 secret_key: str jwt_algorithm: str = "HS256" access_token_expire_minutes: int = 30 # API配置 api_prefix: str = "/api/v1" cors_origins: list = ["http://localhost:3000"] # 第三方服务 sentry_dsn: Optional[str] = None email_host: Optional[str] = None email_port: int = 587 class Config: env_file = ".env" case_sensitive = False @validator("environment", pre=True) def validate_environment(cls, v): """验证环境变量""" if isinstance(v, str): return Environment(v.lower) return v @validator("secret_key") def validate_secret_key(cls, v): """验证密钥""" if len(v) Settings: """获取设置""" return self._settings def reload(self): """重新加载配置""" self._settings = Settings self._load_additional_config def get_database_config(self) -> DatabaseConfig: """获取数据库配置""" return DatabaseConfig( url=self._settings.database_url, echo=self._settings.database_echo ) def get_security_config(self) -> SecurityConfig: """获取安全配置""" return SecurityConfig( secret_key=self._settings.secret_key, algorithm=self._settings.jwt_algorithm, access_token_expire_minutes=self._settings.access_token_expire_minutes )def setup_logging(environment: Environment) -> Dict[str, Any]: """设置日志配置""" base_config = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'standard': { 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s' }, 'json': { 'format': '{"time": "%(asctime)s", "name": "%(name)s", "level": "%(levelname)s", "message": "%(message)s"}' }, }, 'handlers': { 'default': { 'level': 'INFO', 'formatter': 'standard', 'class': 'logging.StreamHandler', 'stream': 'ext://sys.stdout', }, 'file': { 'level': 'INFO', 'formatter': 'standard', 'class': 'logging.handlers.RotatingFileHandler', 'filename': 'logs/taskmaster.log', 'maxBytes': 10485760, # 10MB 'backupCount': 5, }, }, 'loggers': { 'taskmaster': { 'handlers': ['default', 'file'], 'level': 'INFO', 'propagate': False }, }, 'root': { 'level': 'INFO', 'handlers': ['default'] } } # 根据环境调整日志级别 if environment == Environment.DEVELOPMENT: base_config['loggers']['taskmaster']['level'] = 'DEBUG' base_config['root']['level'] = 'DEBUG' elif environment == Environment.PRODUCTION: base_config['handlers']['default']['formatter'] = 'json' base_config['loggers']['taskmaster']['handlers'] = ['file'] return base_configdef demonstrate_config_management: """演示配置管理""" print("\n=== 配置管理与环境设置 ===") # 设置测试环境变量 os.environ['SECRET_KEY'] = 'a-very-long-secret-key-for-testing-purposes-only' os.environ['DATABASE_URL'] = 'sqlite:///./test.db' os.environ['ENVIRONMENT'] = 'development' try: # 初始化配置管理器 config_manager = ConfigManager settings = config_manager.settings print("应用配置:") print(f" 应用名称: {settings.app_name}") print(f" 环境: {settings.environment.value}") print(f" 调试模式: {settings.debug}") print(f" API前缀: {settings.api_prefix}") # 获取数据库配置 db_config = config_manager.get_database_config print(f"\n数据库配置:") print(f" 连接URL: {db_config.url}") print(f" SQL回显: {db_config.echo}") print(f" 连接池大小: {db_config.pool_size}") # 获取安全配置 security_config = config_manager.get_security_config print(f"\n安全配置:") print(f" 算法: {security_config.algorithm}") print(f" Token过期时间: {security_config.access_token_expire_minutes}分钟") # 设置日志 logging_config = setup_logging(settings.environment) dictConfig(logging_config) logger = logging.getLogger('taskmaster.config') logger.info("日志系统初始化完成") # 演示配置验证 print(f"\n配置验证演示:") try: # 测试无效配置 os.environ['SECRET_KEY'] = 'short' invalid_config_manager = ConfigManager except ValueError as e: print(f" 配置验证错误: {e}") # 恢复有效配置 os.environ['SECRET_KEY'] = 'a-very-long-secret-key-for-testing-purposes-only' # 演示配置重载 print(f"\n配置重载演示:") old_debug = settings.debug os.environ['DEBUG'] = 'true' config_manager.reload new_settings = config_manager.settings print(f" 调试模式变化: {old_debug} -> {new_settings.debug}") except Exception as e: print(f"配置管理演示错误: {e}") finally: # 清理环境变量 for key in ['SECRET_KEY', 'DATABASE_URL', 'ENVIRONMENT', 'DEBUG']: if key in os.environ: del os.environ[key]# 运行配置管理演示demonstrate_config_management"""完整的应用启动器和项目管理"""import asyncioimport signalimport sysfrom contextlib import asynccontextmanagerfrom fastapi import FastAPI, Depends, HTTPExceptionfrom fastapi.middleware.cors import CORSMiddlewareimport uvicornclass TaskMasterApplication: """TaskMaster应用主类""" def __init__(self): self.config_manager = ConfigManager self.settings = self.config_manager.settings self.app = None self._setup_complete = False def setup(self) -> 'TaskMasterApplication': """应用设置""" if self._setup_complete: return self # 创建FastAPI应用 @asynccontextmanager async def lifespan(app: FastAPI): # 启动时执行 await self._on_startup yield # 关闭时执行 await self._on_shutdown self.app = FastAPI( title=self.settings.app_name, version="0.1.0", description="现代化的任务管理系统", lifespan=lifespan ) # 添加中间件 self._add_middleware # 设置路由 self._setup_routes # 配置异常处理 self._setup_exception_handlers self._setup_complete = True return self def _add_middleware(self): """添加中间件""" # CORS中间件 self.app.add_middleware( CORSMiddleware, allow_origins=self.settings.cors_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 可以添加更多中间件，如认证、日志等 def _setup_routes(self): """设置路由""" @self.app.get("/") async def root: return { "message": f"欢迎使用 {self.settings.app_name}", "version": "0.1.0", "environment": self.settings.environment.value } @self.app.get("/health") async def health_check: return { "status": "healthy", "timestamp": datetime.now.isoformat } @self.app.get("/config") async def get_config(current_user: dict = Depends(self._get_current_user)): """获取配置信息（需要认证）""" return { "app_name": self.settings.app_name, "environment": self.settings.environment.value, "debug": self.settings.debug, "api_prefix": self.settings.api_prefix } def _setup_exception_handlers(self): """设置异常处理器""" @self.app.exception_handler(HTTPException) async def http_exception_handler(request, exc): return { "error": exc.detail, "status_code": exc.status_code, "path": request.url.path } @self.app.exception_handler(Exception) async def general_exception_handler(request, exc): # 记录错误日志 logger = logging.getLogger('taskmaster.application') logger.error(f"未处理异常: {exc}", exc_info=True) return { "error": "内部服务器错误", "status_code": 500, "path": request.url.path } async def _get_current_user(self, token: str = Depends(lambda: "demo")): """获取当前用户（演示用）""" # 在实际应用中，这里会验证JWT token return {"user_id": 1, "username": "demo_user"} async def _on_startup(self): """应用启动时执行""" logger = logging.getLogger('taskmaster.application') logger.info(f"启动 {self.settings.app_name} - {self.settings.environment.value}") # 初始化数据库连接 # 运行数据库迁移 # 启动后台任务等 print(f" {self.settings.app_name} 启动完成!") print(f" 环境: {self.settings.environment.value}") print(f" 调试模式: {self.settings.debug}") async def _on_shutdown(self): """应用关闭时执行""" logger = logging.getLogger('taskmaster.application') logger.info("应用正在关闭...") # 关闭数据库连接 # 停止后台任务 # 清理资源 print(" 应用已安全关闭") def run(self, host: str = "0.0.0.0", port: int = 8000): """运行应用""" if not self._setup_complete: self.setup uvicorn.run( self.app, host=host, port=port, log_config=setup_logging(self.settings.environment), reload=self.settings.debug )class CLI: """命令行接口""" def __init__(self): self.application = TaskMasterApplication def run(self): """运行CLI""" if len(sys.argv) 可用命令: start 启动开发服务器 test 运行测试 migrate 运行数据库迁移 shell 打开Python shell setup 设置开发环境 --help 显示此帮助信息示例: python -m taskmaster start python -m taskmaster test python -m taskmaster setup""" print(help_text)def demonstrate_project_management: """演示项目管理""" print("\n=== 完整的项目工程化实践 ===") # 演示CLI功能 print("1. 命令行接口演示") cli = CLI # 模拟命令行参数 original_argv = sys.argv try: # 测试帮助命令 sys.argv = ['taskmaster', '--help'] print("\n帮助命令输出:") cli.show_help # 测试环境设置 print("\n环境设置演示:") cli.setup_environment finally: sys.argv = original_argv # 演示应用启动 print("\n2. 应用启动演示") app = TaskMasterApplication.setup print("应用信息:") print(f" 应用名称: {app.settings.app_name}") print(f" 环境: {app.settings.environment.value}") print(f" API文档: http://127.0.0.1:8000/docs") print(f" 健康检查: http://127.0.0.1:8000/health") # 项目质量检查 print("\n3. 项目质量检查") quality_checks = [ ("代码结构", self_check_code_structure), ("配置管理", self_check_config_management), ("文档完整性", self_check_documentation), ("测试覆盖", self_check_testing), ] for check_name, check_func in quality_checks: result, message = check_func status = "✓" if result else "✗" print(f" {status} {check_name}: {message}") print("\n 项目工程化最佳实践总结:") print("✓ 清晰的项目结构和职责分离") print("✓ 环境特定的配置管理") print("✓ 完善的日志系统") print("✓ 健康检查和监控") print("✓ 命令行工具和脚本") print("✓ 容器化部署支持") print("✓ 持续集成配置") print("✓ 代码质量工具集成")def self_check_code_structure: """检查代码结构""" required_dirs = ['taskmaster/src', 'taskmaster/tests', 'docs', 'scripts'] missing_dirs = [d for d in required_dirs if not os.path.exists(d)] if missing_dirs: return False, f"缺少目录: {', '.join(missing_dirs)}" return True, "代码结构完整"def self_check_config_management: """检查配置管理""" required_configs = ['.env.example', 'config/default.py'] missing_configs = [c for c in required_configs if not os.path.exists(c)] if missing_configs: return False, f"缺少配置文件: {', '.join(missing_configs)}" return True, "配置管理完善"def self_check_documentation: """检查文档完整性""" required_docs = ['README.md', 'docs/index.rst'] missing_docs = [d for d in required_docs if not os.path.exists(d)] if missing_docs: return False, f"缺少文档: {', '.join(missing_docs)}" return True, "文档完整"def self_check_testing: """检查测试覆盖""" if not os.path.exists('tests'): return False, "缺少测试目录" test_files = [f for f in os.listdir('tests') if f.startswith('test_')] if not test_files: return False, "缺少测试文件" return True, f"找到 {len(test_files)} 个测试文件"# 运行完整的项目演示demonstrate_project_managementprint("\n" + "="*60)print("第二十二天学习完成！")print("="*60)print("今天学习了完整的Python项目工程化:")print("✓ 标准的项目结构和组织")print("✓ 配置管理和环境设置")print("✓ 日志系统和错误处理")print("✓ 应用生命周期管理")print("✓ 命令行工具开发")print("✓ 容器化和部署配置")print("✓ 项目质量保证")print("✓ 团队协作和文档")项目结构设计：
标准的Python包布局模块化设计和职责分离配置、测试、文档目录组织配置管理：
环境特定的配置安全的敏感信息处理配置验证和类型安全应用架构：
依赖注入和生命周期管理中间件和异常处理API设计和路由管理开发工具链：
命令行接口开发测试框架集成代码质量工具容器化部署文档编写和维护代码审查流程持续集成配置部署流水线来源：琢磨先生起飞吧
