https://www.python.org/static/community_logos/python-logo-master-v3-TM.png

大型项目结构与设计模式

项目结构规范

text

复制

下载

enterprise_app/
├── docs/                  # 项目文档
├── tests/                 # 测试代码
│   ├── unit/              # 单元测试
│   └── integration/       # 集成测试
├── src/                   # 主代码
│   ├── package/           # 主包
│   │   ├── __init__.py
│   │   ├── core/          # 核心业务逻辑
│   │   ├── models/        # 数据模型
│   │   ├── services/      # 服务层
│   │   ├── api/           # API接口
│   │   └── utils/         # 工具函数
│   └── scripts/           # 脚本目录
├── configs/               # 配置文件
├── requirements/          # 依赖文件
│   ├── base.txt           # 基础依赖
│   ├── dev.txt            # 开发依赖
│   └── prod.txt           # 生产依赖
├── .env                   # 环境变量
├── .gitignore
├── pyproject.toml         # 项目配置
└── README.md

工厂模式实现

python

复制

下载

from abc import ABC, abstractmethodclass DatabaseConnection(ABC):@abstractmethoddef connect(self):pass@abstractmethoddef execute_query(self, query):passclass MySQLConnection(DatabaseConnection):def connect(self):print("Connecting to MySQL database")return selfdef execute_query(self, query):print(f"Executing MySQL query: {query}")class PostgreSQLConnection(DatabaseConnection):def connect(self):print("Connecting to PostgreSQL database")return selfdef execute_query(self, query):print(f"Executing PostgreSQL query: {query}")class DatabaseFactory:@staticmethoddef create_connection(db_type):if db_type == "mysql":return MySQLConnection()elif db_type == "postgresql":return PostgreSQLConnection()else:raise ValueError("Unsupported database type")# 使用工厂
db = DatabaseFactory.create_connection("mysql")
db.connect().execute_query("SELECT * FROM users")

https://refactoring.guru/images/patterns/diagrams/factory-method/structure.png

企业级Web框架：Django

Django项目结构

text

复制

下载

django_project/
├── manage.py
├── project/
│   ├── __init__.py
│   ├── asgi.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── apps/├── users/│   ├── migrations/│   ├── __init__.py│   ├── admin.py│   ├── apps.py│   ├── models.py│   ├── tests.py│   ├── urls.py│   └── views.py└── products/├── migrations/├── __init__.py├── admin.py├── apps.py├── models.py├── tests.py├── urls.py└── views.py

Django REST Framework示例

python

复制

下载

# serializers.py
from rest_framework import serializers
from .models import Productclass ProductSerializer(serializers.ModelSerializer):class Meta:model = Productfields = ['id', 'name', 'price', 'description']# views.py
from rest_framework import viewsets
from .models import Product
from .serializers import ProductSerializerclass ProductViewSet(viewsets.ModelViewSet):queryset = Product.objects.all()serializer_class = ProductSerializerfilterset_fields = ['price']search_fields = ['name', 'description']# urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import ProductViewSetrouter = DefaultRouter()
router.register(r'products', ProductViewSet)urlpatterns = [path('', include(router.urls)),
]

https://www.django-rest-framework.org/img/logo.png

容器化与部署

Dockerfile示例

dockerfile

复制

下载

# 使用官方Python基础镜像
FROM python:3.9-slim# 设置工作目录
WORKDIR /app# 设置环境变量
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \build-essential \&& rm -rf /var/lib/apt/lists/*# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt# 复制项目代码
COPY . .# 暴露端口
EXPOSE 8000# 运行命令
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "project.wsgi:application"]

Kubernetes部署配置

yaml

复制

下载

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:name: django-app
spec:replicas: 3selector:matchLabels:app: djangotemplate:metadata:labels:app: djangospec:containers:- name: djangoimage: your-registry/django-app:latestports:- containerPort: 8000envFrom:- configMapRef:name: django-config# service.yaml
apiVersion: v1
kind: Service
metadata:name: django-service
spec:selector:app: djangoports:- protocol: TCPport: 80targetPort: 8000type: LoadBalancer

https://d33wubrfki0l68.cloudfront.net/2475489eaf20163ec0f54ddc1d92aa8d4c87c96b/e7c81/images/docs/components-of-kubernetes.svg

持续集成与部署 (CI/CD)

GitHub Actions配置

yaml

复制

下载

# .github/workflows/ci-cd.yaml
name: CI/CD Pipelineon:push:branches: [ main ]pull_request:branches: [ main ]jobs:test:runs-on: ubuntu-lateststeps:- uses: actions/checkout@v2- name: Set up Pythonuses: actions/setup-python@v2with:python-version: '3.9'- name: Install dependenciesrun: |python -m pip install --upgrade pippip install -r requirements.txtpip install pytest- name: Run testsrun: |pytestdeploy:needs: testruns-on: ubuntu-latestif: github.ref == 'refs/heads/main'steps:- uses: actions/checkout@v2- name: Build Docker imagerun: docker build -t django-app .- name: Log in to Docker Hubrun: echo "${{ secrets.DOCKER_PASSWORD }}" | docker login -u "${{ secrets.DOCKER_USERNAME }}" --password-stdin- name: Push Docker imagerun: |docker tag django-app your-username/django-app:latestdocker push your-username/django-app:latest- name: Deploy to Kubernetesrun: |kubectl apply -f k8s/

https://www.redhat.com/cms/managed-files/ci-cd-flow-desktop-2.png

监控与日志

Prometheus监控配置

python

复制

下载

# prometheus_client示例
from prometheus_client import start_http_server, Counter, Gauge
import random
import time# 定义指标
REQUEST_COUNT = Counter('app_requests_total', 'Total HTTP Requests')
TEMPERATURE = Gauge('app_temperature_celsius', 'Current temperature')def process_request():REQUEST_COUNT.inc()TEMPERATURE.set(random.uniform(18.0, 25.0))if __name__ == '__main__':# 启动指标服务器start_http_server(8000)# 模拟请求while True:process_request()time.sleep(2)

ELK日志收集配置

python

复制

下载

# logging配置示例
import logging
from pythonjsonlogger import jsonloggerdef setup_logging():logger = logging.getLogger()logger.setLevel(logging.INFO)# JSON格式化formatter = jsonlogger.JsonFormatter('%(asctime)s %(levelname)s %(name)s %(message)s')# 控制台处理器console_handler = logging.StreamHandler()console_handler.setFormatter(formatter)logger.addHandler(console_handler)# 文件处理器file_handler = logging.FileHandler('app.log')file_handler.setFormatter(formatter)logger.addHandler(file_handler)return loggerlogger = setup_logging()
logger.info("Application started", extra={"user": "admin", "module": "startup"})

https://www.elastic.co/guide/en/elasticsearch/reference/current/images/elas_0201.png

微服务架构

FastAPI微服务示例

python

复制

下载

from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strprice: floatis_offer: bool = None@app.get("/")
def read_root():return {"Hello": "World"}@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):return {"item_id": item_id, "q": q}@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):return {"item_name": item.name, "item_id": item_id}# 运行: uvicorn main:app --reload

服务间通信

python

复制

下载

# 使用requests同步调用
import requestsdef get_user_data(user_id):response = requests.get(f"http://user-service/users/{user_id}",timeout=3)response.raise_for_status()return response.json()# 使用aiohttp异步调用
import aiohttpasync def async_get_user_data(user_id):async with aiohttp.ClientSession() as session:async with session.get(f"http://user-service/users/{user_id}") as response:return await response.json()

https://microservices.io/i/architecture.png

安全最佳实践

JWT认证实现

python

复制

下载

from datetime import datetime, timedelta
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearerSECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")def create_access_token(data: dict):to_encode = data.copy()expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp": expire})encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)return encoded_jwtdef verify_token(token: str = Depends(oauth2_scheme)):try:payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])return payloadexcept jwt.PyJWTError:raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Invalid authentication credentials",headers={"WWW-Authenticate": "Bearer"},)# 保护路由
@app.get("/protected")
async def protected_route(payload: dict = Depends(verify_token)):return {"message": "Access granted", "user": payload.get("sub")}

安全头部中间件

python

复制

下载

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZipMiddlewareapp = FastAPI()# 强制HTTPS
app.add_middleware(HTTPSRedirectMiddleware)# 可信主机
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com"])# 安全头部
@app.middleware("http")
async def add_security_headers(request, call_next):response = await call_next(request)response.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains"response.headers["X-Content-Type-Options"] = "nosniff"response.headers["X-Frame-Options"] = "DENY"response.headers["X-XSS-Protection"] = "1; mode=block"response.headers["Content-Security-Policy"] = "default-src 'self'"return response

性能调优

数据库优化

python

复制

下载

# 不良实践 - N+1查询问题
users = User.objects.all()
for user in users:print(user.profile.bio)  # 每次循环都查询数据库# 优化方案 - select_related/prefetch_related
users = User.objects.select_related('profile').all()
for user in users:print(user.profile.bio)  # 仅一次查询# 使用索引
class User(models.Model):name = models.CharField(max_length=100, db_index=True)email = models.EmailField(unique=True)class Meta:indexes = [models.Index(fields=['name', 'email']),]

缓存策略

python

复制

下载

from django.core.cache import cachedef get_expensive_data():# 尝试从缓存获取data = cache.get("expensive_data")if data is None:# 缓存未命中，计算数据data = calculate_expensive_data()# 设置缓存，有效期1小时cache.set("expensive_data", data, timeout=3600)return data# Redis缓存后端配置
CACHES = {"default": {"BACKEND": "django_redis.cache.RedisCache","LOCATION": "redis://127.0.0.1:6379/1","OPTIONS": {"CLIENT_CLASS": "django_redis.client.DefaultClient",}}
}

企业级测试策略

测试金字塔实现

python

复制

下载

# 单元测试示例
import unittest
from unittest.mock import Mock, patchclass TestUserService(unittest.TestCase):@patch('services.user_service.UserRepository')def test_create_user(self, mock_repo):mock_repo.return_value.save.return_value = {"id": 1, "name": "test"}result = UserService().create_user("test")self.assertEqual(result["name"], "test")# 集成测试示例
from django.test import TestCaseclass UserAPITest(TestCase):def test_user_creation(self):response = self.client.post('/api/users/', {'name': 'test'})self.assertEqual(response.status_code, 201)self.assertEqual(response.json()['name'], 'test')# E2E测试示例
from selenium import webdriverclass UserJourneyTest(unittest.TestCase):def setUp(self):self.driver = webdriver.Chrome()def test_user_registration(self):self.driver.get("http://localhost:8000/register")self.driver.find_element_by_id("name").send_keys("test")self.driver.find_element_by_id("submit").click()self.assertIn("Welcome", self.driver.page_source)def tearDown(self):self.driver.quit()

https://martinfowler.com/articles/practical-test-pyramid/test-pyramid.png