Python Web开发：使用FastAPI构建社交网络系统

Python Web开发：使用FastAPI构建社交网络系统

目录

• 🌐 用户管理与消息系统
• 💬 点赞、评论、关注功能
• 📢 实时通知系统（WebSocket）

1. 用户管理与消息系统

用户管理是任何社交网络系统的基础。通过对用户的身份进行验证、信息存储和管理，可以保证平台的安全性与用户体验的流畅度。FastAPI作为一个高性能的Web框架，能够为社交网络系统提供高效的API接口，完成从用户注册、登录到消息交流等一系列功能。

用户注册与登录

用户注册是每个社交平台的第一步。为了确保用户信息的安全性，我们首先需要对用户输入的数据进行验证，并对密码进行加密存储。密码的存储采用哈希加密，可以使用如bcrypt、pbkdf2等加密算法。以下是一个完整的用户注册与登录功能实现代码：

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from passlib.context import CryptContext
from pydantic import BaseModel
from datetime import datetime, timedelta
import jwt

app = FastAPI()
db_session = Session()

class User(BaseModel):
    username: str
    email: str
    password: str

# 密码加密
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def get_password_hash(password: str):
    return pwd_context.hash(password)

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

class Token(BaseModel):
    access_token: str
    token_type: str

# 用户注册
@app.post("/register/")
def register(user: User, db: Session = Depends(db_session)):
    existing_user = db.query(UserModel).filter(UserModel.username == user.username).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Username already registered")
    hashed_password = get_password_hash(user.password)
    new_user = UserModel(username=user.username, email=user.email, password=hashed_password)
    db.add(new_user)
    db.commit()
    return {"message": "User registered successfully"}

# 用户登录
@app.post("/login/", response_model=Token)
def login(user: User, db: Session = Depends(db_session)):
    existing_user = db.query(UserModel).filter(UserModel.username == user.username).first()
    if not existing_user or not verify_password(user.password, existing_user.password):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    access_token = jwt.encode({"sub": user.username, "exp": datetime.utcnow() + timedelta(hours=1)}, "secret", algorithm="HS256")
    return {"access_token": access_token, "token_type": "bearer"}

• 在上述代码中，首先定义了User和Token两个模型，分别用于接收注册信息和返回登录令牌。
• 在register接口中，检查用户是否已存在，若不存在则对密码进行哈希加密后存储至数据库。
• login接口根据提供的用户名和密码验证用户身份，通过生成JWT令牌来保证用户后续的身份认证。

这种做法不仅提高了用户数据的安全性，也能大大简化后端认证逻辑。

消息系统实现

在社交网络中，用户间的互动非常重要，消息系统是平台互动的一大关键功能。通过消息系统，用户能够发送私信、查看收到的消息以及对消息进行管理。FastAPI能够利用ORM与数据库结合，轻松实现消息的存储与检索。

为了实现高效的消息管理，我们首先需要设计消息表结构，并提供用户发送消息和查看消息的接口。以下是一个完整的消息系统代码示例：

class Message(BaseModel):
    sender_id: int
    receiver_id: int
    content: str
    timestamp: datetime

# 发送消息
@app.post("/send_message/")
def send_message(message: Message, db: Session = Depends(db_session)):
    new_message = MessageModel(sender_id=message.sender_id, receiver_id=message.receiver_id, content=message.content, timestamp=datetime.utcnow())
    db.add(new_message)
    db.commit()
    return {"message": "Message sent successfully"}

# 获取用户消息
@app.get("/get_messages/{user_id}")
def get_messages(user_id: int, db: Session = Depends(db_session)):
    messages = db.query(MessageModel).filter(MessageModel.receiver_id == user_id).all()
    return messages

• MessageModel代表数据库中的消息表，存储每条消息的发送者、接收者、内容以及发送时间。
• 发送消息时，系统会将消息保存到数据库中并返回成功信息。
• 用户获取消息时，系统会查询数据库并返回该用户收到的所有消息。

这种设计可以确保用户之间的消息交流顺畅，并且能够在未来根据需求进行功能扩展，例如添加消息删除、消息标记已读等功能。

2. 点赞、评论、关注功能

点赞、评论、关注是社交网络中最常见的互动功能。通过这些功能，用户能够更好地与平台内容进行互动，增强用户粘性和活跃度。FastAPI能够简洁高效地实现这些功能，并且支持与数据库中的内容进行关联。

点赞功能

点赞是最直接的互动行为，通常与内容（如帖子、图片或评论）绑定。每当用户对某一内容进行点赞时，系统需要记录该用户的点赞行为。为了避免重复点赞，可以在设计中引入检查机制。

以下是点赞功能的实现代码：

class Like(BaseModel):
    user_id: int
    content_id: int

# 点赞
@app.post("/like/")
def like(content: Like, db: Session = Depends(db_session)):
    existing_like = db.query(LikeModel).filter(LikeModel.user_id == content.user_id, LikeModel.content_id == content.content_id).first()
    if existing_like:
        raise HTTPException(status_code=400, detail="You already liked this content")
    new_like = LikeModel(user_id=content.user_id, content_id=content.content_id)
    db.add(new_like)
    db.commit()
    return {"message": "Liked successfully"}

• 该功能中，首先定义了一个LikeModel来记录点赞行为。
• 点赞时，系统会检查用户是否已对该内容点赞，若未点赞，则将点赞信息存入数据库。

点赞功能能够帮助平台对用户的偏好进行统计，为后续的推荐算法或内容展示提供有力支持。

评论功能

评论功能允许用户对帖子、图片等内容进行评论。每条评论都需要记录评论者的信息、评论的内容以及评论的时间戳。这使得评论系统不仅是一个互动功能，也为平台提供了丰富的用户生成内容（UGC）。

以下是评论功能的实现代码：

class Comment(BaseModel):
    user_id: int
    content_id: int
    text: str

# 添加评论
@app.post("/comment/")
def add_comment(comment: Comment, db: Session = Depends(db_session)):
    new_comment = CommentModel(user_id=comment.user_id, content_id=comment.content_id, text=comment.text, timestamp=datetime.utcnow())
    db.add(new_comment)
    db.commit()
    return {"message": "Comment added successfully"}

• 每条评论记录的结构包含了评论者的ID、评论内容ID、评论文本以及时间戳。
• 用户发表评论时，系统会将评论信息存入数据库并返回成功消息。

评论功能的引入不仅增强了内容互动，还使得平台能够建立起内容的讨论区，进一步促进了用户参与。

关注功能

关注是社交平台中的核心社交功能之一。通过关注功能，用户可以关注其他用户并及时获取他们的动态。这一功能通常通过一个“关注”关系表来实现，每一条记录都表示一个用户对另一个用户的关注关系。

以下是关注功能的实现代码：

class Follow(BaseModel):
    follower_id: int
    following_id: int

# 关注用户
@app.post("/follow/")
def follow(follow: Follow, db: Session = Depends(db_session)):
    existing_follow = db.query(FollowModel).filter(FollowModel.follower_id == follow.follower_id, FollowModel.following_id == follow.following_id).first()
    if existing_follow:
        raise HTTPException(status_code=400, detail="You already follow this user")
    new_follow = FollowModel(follower_id=follow.follower_id, following_id=follow.following_id)
    db.add(new_follow)
    db.commit()
    return {"message": "Followed successfully"}

• 该功能记录了用户与其他用户之间的关注关系。
• 每当用户关注某人时，系统会在FollowModel表中创建一条新的记录，标记出关注者和被关注者。

关注功能能够有效地增强平台的社交属性，帮助用户发现更多感兴趣的内容和用户。

3. 实时通知系统（WebSocket）

在社交平台中，用户的互动（如新消息、新评论、点赞等）往往需要实时通知给用户。传统的HTTP请求方式无法及时推送这些信息，而WebSocket技术则能够实现双向实时通信，解决这一问题。FastAPI对WebSocket的支持，使得实时通知功能的实现变得简单高效。

WebSocket推送通知

WebSocket建立了一个持久化的连接，通过该连接，服务器可以主动推送信息到客户端。这对于实时消息推送、评论通知、点赞通知等社交网络功能非常有用。

以下是WebSocket通知功能的实现代码：

from fastapi import WebSocket, WebSocketDisconnect

active_connections = {}

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: int):
    await websocket.accept()
    active_connections[user_id] = websocket
    try:
        while True:
            data = await websocket.receive_text()
            # 在此可以处理来自客户端的消息
            print(f"Message from {user_id}: {data}")
    except WebSocketDisconnect:
        del active_connections[user_id]
        print(f"User {user_id} disconnected")
        
# 推送通知
async def send_notification(user_id: int, message: str):
    if user_id in active_connections:
        await active_connections[user_id].send_text(message)

• 在/ws/{user_id}路径下，我们使用WebSocket协议建立与客户端的连接。
• 当服务器端有新的通知时，它能够通过WebSocket连接将通知实时推送到客户端。

这种设计方式极大提升了用户体验，使得平台能够即时通知用户关于新消息、点赞、评论等事件，确保平台的互动性与活跃度。