Python 企業級開發與DevOps實踐

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