Python后端开发最佳实践与常用框架对比#

1
# Django项目结构示例
2
myproject/
3
├── manage.py
4
├── myproject/
5
│   ├── __init__.py
6
│   ├── settings.py
7
│   ├── urls.py
8
│   └── wsgi.py
9
└── apps/
10
    ├── users/
11
    ├── blog/
12
    └── api/

1
flask-app/
2
├── app/
3
│   ├── __init__.py
4
│   ├── models.py
5
│   ├── views.py
6
│   └── utils.py
7
├── migrations/
8
├── tests/
9
├── config.py
10
└── run.py

1
from fastapi import FastAPI, HTTPException
2
from pydantic import BaseModel
3
from typing import List, Optional
4

5
app = FastAPI(title="My API", version="1.0.0")
6

7
class User(BaseModel):
8
    id: int
9
    name: str
10
    email: str
11
    age: Optional[int] = None
12

13
@app.get("/users", response_model=List[User])
14
async def get_users():
15
    return await fetch_users_from_db()
16

17
@app.post("/users", response_model=User)
18
async def create_user(user: User):
19
    return await save_user_to_db(user)

1
from databases import Database
2
from sqlalchemy import create_engine, MetaData
3

4
database = Database(DATABASE_URL)
5
metadata = MetaData()
6

7
@app.on_event("startup")
8
async def startup():
9
    await database.connect()
10

11
@app.on_event("shutdown")
12
async def shutdown():
13
    await database.disconnect()
14

15
async def fetch_users_from_db():
16
    query = "SELECT * FROM users"
17
    return await database.fetch_all(query)

1
backend/
2
├── app/
3
│   ├── api/          # API路由层
4
│   ├── core/         # 核心配置
5
│   ├── crud/         # 数据访问层
6
│   ├── db/           # 数据库配置
7
│   ├── models/       # 数据模型
8
│   ├── schemas/      # 数据验证模式
9
│   └── services/     # 业务逻辑层
10
├── tests/
11
├── alembic/          # 数据库迁移
12
└── requirements.txt

1
from sqlalchemy import create_engine
2
from sqlalchemy.pool import QueuePool
3

4
engine = create_engine(
5
    DATABASE_URL,
6
    poolclass=QueuePool,
7
    pool_size=20,
8
    max_overflow=0,
9
    pool_pre_ping=True,
10
    pool_recycle=300
11
)

1
# 使用索引
2
class User(Base):
3
    __tablename__ = "users"
4

5
    id = Column(Integer, primary_key=True)
6
    email = Column(String, unique=True, index=True)
7
    created_at = Column(DateTime, default=datetime.utcnow, index=True)
8

9
# 批量操作
10
def bulk_create_users(users_data):
11
    users = [User(**data) for data in users_data]
12
    db.bulk_save_objects(users)
13
    db.commit()

1
# 资源路由设计
2
@app.get("/api/v1/users")           # 获取用户列表
3
@app.post("/api/v1/users")          # 创建用户
4
@app.get("/api/v1/users/{user_id}") # 获取单个用户
5
@app.put("/api/v1/users/{user_id}") # 更新用户
6
@app.delete("/api/v1/users/{user_id}") # 删除用户

1
from pydantic import BaseModel
2
from typing import Any, Optional
3

4
class APIResponse(BaseModel):
5
    success: bool
6
    message: str
7
    data: Optional[Any] = None
8
    error_code: Optional[str] = None
9

10
@app.get("/users")
11
async def get_users():
12
    try:
13
        users = await fetch_users()
14
        return APIResponse(
15
            success=True,
16
            message="获取用户列表成功",
17
            data=users
18
        )
19
    except Exception as e:
20
        return APIResponse(
21
            success=False,
22
            message="获取用户列表失败",
23
            error_code="USER_FETCH_ERROR"
24
        )

1
from jose import JWTError, jwt
2
from passlib.context import CryptContext
3

4
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
5

6
def verify_password(plain_password, hashed_password):
7
    return pwd_context.verify(plain_password, hashed_password)
8

9
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
10
    to_encode = data.copy()
11
    if expires_delta:
12
        expire = datetime.utcnow() + expires_delta
13
    else:
14
        expire = datetime.utcnow() + timedelta(minutes=15)
15

16
    to_encode.update({"exp": expire})
17
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
18
    return encoded_jwt

1
from pydantic import BaseModel, validator, EmailStr
2

3
class UserCreate(BaseModel):
4
    email: EmailStr
5
    password: str
6
    age: int
7

8
    @validator('password')
9
    def validate_password(cls, v):
10
        if len(v) < 8:
11
            raise ValueError('密码长度至少8位')
12
        if not any(char.isdigit() for char in v):
13
            raise ValueError('密码必须包含数字')
14
        return v
15

16
    @validator('age')
17
    def validate_age(cls, v):
18
        if v < 0 or v > 150:
19
            raise ValueError('年龄必须在0-150之间')
20
        return v

1
import redis
2
from functools import wraps
3
import json
4

5
redis_client = redis.Redis(host='localhost', port=6379, db=0)
6

7
def cache_result(expire_time=300):
8
    def decorator(func):
9
        @wraps(func)
10
        async def wrapper(*args, **kwargs):
11
            # 生成缓存key
12
            cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
13

14
            # 尝试从缓存获取
15
            cached_result = redis_client.get(cache_key)
16
            if cached_result:
17
                return json.loads(cached_result)
18

19
            # 执行函数并缓存结果
20
            result = await func(*args, **kwargs)
21
            redis_client.setex(
22
                cache_key,
23
                expire_time,
24
                json.dumps(result, default=str)
25
            )
26
            return result
27
        return wrapper
28
    return decorator
29

30
@cache_result(expire_time=600)
31
async def get_user_profile(user_id: int):
32
    return await fetch_user_from_db(user_id)