本文目录导读：
我来为你详细介绍Python项目迁移工具的封装方法和实际案例。
基础迁移工具封装结构
# migration_tool.py
import os
import sys
import shutil
import json
import yaml
from pathlib import Path
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class MigrationConfig:
"""迁移配置类"""
source_dir: str
target_dir: str
project_name: str
include_patterns: List[str] = None
exclude_patterns: List[str] = None
env_mapping: Dict[str, str] = None
config_transform: Dict[str, str] = None
def __post_init__(self):
if self.include_patterns is None:
self.include_patterns = ['*.py', '*.json', '*.yaml', '*.yml']
if self.exclude_patterns is None:
self.exclude_patterns = ['__pycache__', '*.pyc', '.git']
if self.env_mapping is None:
self.env_mapping = {}
if self.config_transform is None:
self.config_transform = {}
核心迁移引擎
# migration_engine.py
import re
import fnmatch
from pathlib import Path
class MigrationEngine:
"""迁移引擎"""
def __init__(self, config: MigrationConfig):
self.config = config
self.migration_log = []
def analyze_project(self) -> Dict:
"""分析项目结构"""
project_info = {
'name': self.config.project_name,
'files': [],
'dependencies': [],
'structure': {}
}
source_path = Path(self.config.source_dir)
for file_path in source_path.rglob('*'):
if self._should_include(file_path):
relative_path = file_path.relative_to(source_path)
file_info = {
'path': str(relative_path),
'size': file_path.stat().st_size,
'extension': file_path.suffix,
'modified': datetime.fromtimestamp(
file_path.stat().st_mtime
)
}
project_info['files'].append(file_info)
# 检测依赖
if file_path.suffix == '.py':
deps = self._extract_dependencies(file_path)
project_info['dependencies'].extend(deps)
return project_info
def migrate(self) -> bool:
"""执行迁移"""
try:
# 1. 创建目标目录
target_path = Path(self.config.target_dir)
target_path.mkdir(parents=True, exist_ok=True)
# 2. 创建项目结构
project_root = target_path / self.config.project_name
project_root.mkdir(exist_ok=True)
# 3. 复制并转换文件
source_path = Path(self.config.source_dir)
for file_path in source_path.rglob('*'):
if self._should_include(file_path) and file_path.is_file():
self._migrate_file(file_path, source_path, project_root)
# 4. 生成迁移报告
self._generate_report()
logger.info(f"Migration completed: {self.config.source_dir} -> {project_root}")
return True
except Exception as e:
logger.error(f"Migration failed: {e}")
return False
def _should_include(self, file_path: Path) -> bool:
"""检查文件是否应该包含在迁移中"""
# 检查排除模式
for pattern in self.config.exclude_patterns:
if file_path.match(pattern) or pattern in str(file_path):
return False
# 检查包含模式
if self.config.include_patterns:
for pattern in self.config.include_patterns:
if file_path.match(pattern):
return True
return False
return True
def _migrate_file(self, file_path: Path, source_root: Path, target_root: Path):
"""迁移单个文件"""
relative_path = file_path.relative_to(source_root)
target_path = target_root / relative_path
# 创建目标目录
target_path.parent.mkdir(parents=True, exist_ok=True)
# 根据文件类型处理
if file_path.suffix == '.py':
self._migrate_python_file(file_path, target_path)
elif file_path.suffix in ['.json', '.yaml', '.yml']:
self._migrate_config_file(file_path, target_path)
else:
# 直接复制
shutil.copy2(file_path, target_path)
self.migration_log.append({
'source': str(file_path),
'target': str(target_path),
'action': 'migrated'
})
def _migrate_python_file(self, source: Path, target: Path):
"""迁移Python文件并转换"""
content = source.read_text(encoding='utf-8')
# 应用环境变量映射
content = self._apply_env_mapping(content)
# 应用配置转换
content = self._apply_config_transform(content)
# 写入目标文件
target.write_text(content, encoding='utf-8')
def _migrate_config_file(self, source: Path, target: Path):
"""迁移配置文件"""
if source.suffix == '.json':
with open(source, 'r') as f:
config = json.load(f)
else:
with open(source, 'r') as f:
config = yaml.safe_load(f)
# 应用配置转换
config = self._transform_config(config)
# 写入目标文件
with open(target, 'w') as f:
if target.suffix == '.json':
json.dump(config, f, indent=2)
else:
yaml.dump(config, f)
def _apply_env_mapping(self, content: str) -> str:
"""应用环境变量映射"""
for old_env, new_env in self.config.env_mapping.items():
content = content.replace(old_env, new_env)
return content
def _apply_config_transform(self, content: str) -> str:
"""应用配置转换"""
for old_path, new_path in self.config.config_transform.items():
content = content.replace(old_path, new_path)
return content
def _transform_config(self, config: Dict) -> Dict:
"""转换配置内容"""
if isinstance(config, dict):
transformed = {}
for key, value in config.items():
new_key = self.config.config_transform.get(key, key)
if isinstance(value, dict):
transformed[new_key] = self._transform_config(value)
else:
transformed[new_key] = value
return transformed
return config
def _extract_dependencies(self, file_path: Path) -> List[str]:
"""提取Python文件依赖"""
dependencies = []
content = file_path.read_text(encoding='utf-8')
# 匹配import语句
import_patterns = [
r'^import\s+(\S+)',
r'^from\s+(\S+)\s+import',
]
for pattern in import_patterns:
matches = re.finditer(pattern, content, re.MULTILINE)
for match in matches:
dependency = match.group(1).split('.')[0]
if dependency not in ['os', 'sys', 're', 'json']:
dependencies.append(dependency)
return list(set(dependencies))
def _generate_report(self):
"""生成迁移报告"""
report_file = Path(self.config.target_dir) / f"migration_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
report = {
'project': self.config.project_name,
'timestamp': datetime.now().isoformat(),
'source': self.config.source_dir,
'target': str(Path(self.config.target_dir) / self.config.project_name),
'migrated_files': len(self.migration_log),
'files': self.migration_log
}
with open(report_file, 'w') as f:
json.dump(report, f, indent=2, default=str)
特定的迁移案例
1 Django项目迁移
# django_migration.py
class DjangoMigrationTool:
"""Django项目迁移工具"""
def __init__(self, source_project: str, target_project: str):
self.source_project = source_project
self.target_project = target_project
def migrate_settings(self) -> Dict:
"""迁移Django设置"""
# 读取源设置
sys.path.insert(0, os.path.dirname(self.source_project))
from source.settings import DATABASES, INSTALLED_APPS, MIDDLEWARE
# 创建新的设置
new_settings = {
'DATABASES': self._transform_database_config(DATABASES),
'INSTALLED_APPS': self._update_installed_apps(INSTALLED_APPS),
'MIDDLEWARE': self._update_middleware(MIDDLEWARE)
}
return new_settings
def _transform_database_config(self, databases: Dict) -> Dict:
"""转换数据库配置"""
transformed = {}
for db_name, db_config in databases.items():
if db_config['ENGINE'] == 'django.db.backends.postgresql':
# 转换为PostgreSQL配置
transformed[db_name] = {
'ENGINE': 'django.db.backends.postgresql',
'NAME': os.getenv(f'{db_name.upper()}_NAME'),
'USER': os.getenv(f'{db_name.upper()}_USER'),
'PASSWORD': os.getenv(f'{db_name.upper()}_PASSWORD'),
'HOST': os.getenv(f'{db_name.upper()}_HOST', 'localhost'),
'PORT': os.getenv(f'{db_name.upper()}_PORT', '5432')
}
else:
# 保持原样
transformed[db_name] = db_config
return transformed
def _update_installed_apps(self, apps: List) -> List:
"""更新已安装的应用"""
# 添加新的应用
new_apps = [
'corsheaders',
'rest_framework',
'drf_yasg',
]
apps = list(apps) + new_apps
return list(set(apps))
def _update_middleware(self, middleware: List) -> List:
"""更新中间件"""
# 添加CORS中间件
if 'corsheaders.middleware.CorsMiddleware' not in middleware:
middleware = ['corsheaders.middleware.CorsMiddleware'] + list(middleware)
return middleware
2 Flask微服务迁移
# flask_migration.py
class FlaskMigrationTool:
"""Flask微服务迁移工具"""
def __init__(self, source_dir: str):
self.source_dir = Path(source_dir)
def migrate_to_fastapi(self, output_dir: str) -> Dict:
"""迁移Flask应用到FastAPI"""
migration_results = {
'endpoints_migrated': [],
'models_converted': [],
'errors': []
}
# 扫描所有路由文件
for route_file in self.source_dir.rglob('*routes*.py'):
flask_routes = self._parse_flask_routes(route_file)
fastapi_routes = self._convert_to_fastapi(flask_routes)
# 生成FastAPI路由文件
output_path = Path(output_dir) / route_file.name.replace('routes', 'api')
self._write_fastapi_routes(output_path, fastapi_routes)
migration_results['endpoints_migrated'].extend(
[r['path'] for r in fastapi_routes]
)
return migration_results
def _parse_flask_routes(self, file_path: Path) -> List[Dict]:
"""解析Flask路由"""
routes = []
content = file_path.read_text()
# 匹配Flask路由装饰器
route_pattern = r'@\w+\.route\(\'([^\']+)\'(?:,\s*methods=\[([^\]]+)\])?\)'
matches = re.finditer(route_pattern, content)
for match in matches:
route = {
'path': match.group(1),
'methods': eval(match.group(2)) if match.group(2) else ['GET'],
'function': None # 需要进一步解析
}
routes.append(route)
return routes
def _convert_to_fastapi(self, flask_routes: List[Dict]) -> List[Dict]:
"""转换Flask路由到FastAPI格式"""
fastapi_routes = []
for route in flask_routes:
# 转换路由路径格式
fastapi_path = route['path'].replace('<', '{').replace('>', '}')
fastapi_route = {
'path': fastapi_path,
'methods': route['methods'],
'decorator': self._generate_fastapi_decorator(fastapi_path, route['methods'])
}
fastapi_routes.append(fastapi_route)
return fastapi_routes
def _generate_fastapi_decorator(self, path: str, methods: List[str]) -> str:
"""生成FastAPI装饰器"""
method_decorators = []
for method in methods:
method_lower = method.lower()
if method_lower in ['get', 'post', 'put', 'delete', 'patch']:
method_decorators.append(f"@app.{method_lower}('{path}')")
return '\n'.join(method_decorators)
3 配置迁移示例
# config_migration.py
class ConfigMigrationTool:
"""配置迁移工具"""
def __init__(self, config_path: str):
self.config_path = Path(config_path)
def migrate_ini_to_yaml(self) -> str:
"""迁移INI配置到YAML"""
import configparser
# 读取INI配置
config = configparser.ConfigParser()
config.read(self.config_path)
# 转换为字典
config_dict = {section: dict(config[section])
for section in config.sections()}
# 转换为YAML
yaml_path = self.config_path.with_suffix('.yaml')
with open(yaml_path, 'w') as f:
yaml.dump(config_dict, f, default_flow_style=False)
return str(yaml_path)
def migrate_env_to_dotenv(self, env_file: str) -> str:
"""迁移环境变量到.env文件"""
dotenv_path = Path('.env')
with open(env_file, 'r') as f:
content = f.read()
# 转换格式
transformed = self._transform_env_content(content)
with open(dotenv_path, 'w') as f:
f.write(transformed)
return str(dotenv_path)
def _transform_env_content(self, content: str) -> str:
"""转换环境变量内容"""
lines = content.split('\n')
transformed_lines = []
for line in lines:
line = line.strip()
if line.startswith('#') or not line:
transformed_lines.append(line)
continue
# 确保格式为 KEY=VALUE
if '=' in line:
key, value = line.split('=', 1)
value = value.strip("'\"")
transformed_lines.append(f"{key.strip()}={value}")
return '\n'.join(transformed_lines)
使用示例
# main.py
def main():
"""主函数示例"""
# 1. 基本项目迁移
config = MigrationConfig(
source_dir='/path/to/source/project',
target_dir='/path/to/target',
project_name='my_migrated_project',
env_mapping={
'OLD_DB_HOST': 'NEW_DB_HOST',
'OLD_API_KEY': 'NEW_API_KEY'
},
config_transform={
'old_config_folder': 'new_config_folder',
'legacy_module': 'modern_module'
}
)
engine = MigrationEngine(config)
# 分析项目
analysis = engine.analyze_project()
print(f"Found {len(analysis['files'])} files")
print(f"Dependencies: {analysis['dependencies']}")
# 执行迁移
if engine.migrate():
print("Migration successful!")
else:
print("Migration failed!")
# 2. Django项目迁移
django_migrator = DjangoMigrationTool(
source_project='/path/to/old/django',
target_project='/path/to/new/django'
)
settings = django_migrator.migrate_settings()
# 3. Flask到FastAPI迁移
flask_migrator = FlaskMigrationTool('/path/to/flask/app')
results = flask_migrator.migrate_to_fastapi('/path/to/fastapi/app')
# 4. 配置文件迁移
config_migrator = ConfigMigrationTool('config.ini')
yaml_path = config_migrator.migrate_ini_to_yaml()
if __name__ == "__main__":
main()
高级功能扩展
# advanced_migration.py
class AdvancedMigrationTool:
"""高级迁移工具扩展"""
def __init__(self):
self.transformers = {}
self.validators = []
def register_transformer(self, name: str, transformer_func):
"""注册转换器"""
self.transformers[name] = transformer_func
def register_validator(self, validator_func):
"""注册验证器"""
self.validators.append(validator_func)
def validate_migration(self, source_dir: str, target_dir: str) -> List[str]:
"""验证迁移结果"""
issues = []
for validator in self.validators:
result = validator(source_dir, target_dir)
if result:
issues.extend(result)
return issues
def rollback_migration(self, migration_id: str) -> bool:
"""回滚迁移"""
# 读取迁移日志
log_path = Path(f"migration_{migration_id}.log")
if not log_path.exists():
logger.error(f"Migration log not found: {migration_id}")
return False
# 执行回滚
with open(log_path, 'r') as f:
migration_log = json.load(f)
for action in reversed(migration_log):
if action['action'] == 'migrated':
# 删除迁移的文件
Path(action['target']).unlink(missing_ok=True)
return True
最佳实践建议
1 错误处理
# error_handling.py
class MigrationError(Exception):
"""迁移基础异常"""
pass
class ConfigError(MigrationError):
"""配置异常"""
pass
class FileAccessError(MigrationError):
"""文件访问异常"""
pass
def safe_migration(func):
"""安全迁移装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logger.error(f"Migration error in {func.__name__}: {e}")
raise MigrationError(f"Migration failed: {e}")
return wrapper
2 日志和监控
# logging_config.py
import logging.config
LOGGING_CONFIG = {
'version': 1,
'formatters': {
'detailed': {
'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'detailed',
'level': 'INFO'
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': 'migration.log',
'maxBytes': 1024*1024*5, # 5MB
'backupCount': 5,
'formatter': 'detailed'
}
},
'root': {
'handlers': ['console', 'file'],
'level': 'INFO'
}
}
logging.config.dictConfig(LOGGING_CONFIG)
这个迁移工具框架提供了完整的项目迁移功能,包括：
- 模块化设计：分离配置、引擎和特定迁移逻辑
- 类型安全：使用数据类和类型提示
- 可扩展性：支持自定义转换器和验证器
- 错误处理：完善的异常处理机制
- 日志记录：详细的迁移日志
- 回滚支持：可以撤销迁移操作
- 特定框架支持：Django、Flask等框架的特定迁移逻辑
你可以根据实际需求选择合适的迁移工具进行封装和扩展。
