IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python案例如何搭建接口服务？

wen python案例 12

本文目录导读：

Python案例如何搭建接口服务？

  1. Flask - 轻量级框架
  2. FastAPI - 现代高性能框架
  3. Django REST Framework - 全功能框架
  4. 完整的API服务示例（FastAPI + 数据库）
  5. 测试API服务
  6. 框架选择建议
  7. 部署建议

我来详细介绍如何使用Python搭建接口服务,包含几种常见方案：

Flask - 轻量级框架

基础示例

from flask import Flask, jsonify, request
app = Flask(__name__)
# 模拟数据库
users = [
    {"id": 1, "name": "张三", "age": 25},
    {"id": 2, "name": "李四", "age": 30}
]
# GET请求 - 获取所有用户
@app.route('/api/users', methods=['GET'])
def get_users():
    return jsonify(users)
# GET请求 - 获取单个用户
@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = next((u for u in users if u['id'] == user_id), None)
    if user:
        return jsonify(user)
    return jsonify({"error": "用户不存在"}), 404
# POST请求 - 创建用户
@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    new_user = {
        "id": len(users) + 1,
        "name": data.get('name'),
        "age": data.get('age')
    }
    users.append(new_user)
    return jsonify(new_user), 201
# PUT请求 - 更新用户
@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    data = request.get_json()
    for user in users:
        if user['id'] == user_id:
            user.update(data)
            return jsonify(user)
    return jsonify({"error": "用户不存在"}), 404
if __name__ == '__main__':
    app.run(debug=True, port=5000)

安装与运行

pip install flask
python app.py

FastAPI - 现代高性能框架

基础示例

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# 数据模型
class User(BaseModel):
    id: int
    name: str
    age: int
    email: Optional[str] = None
class UserCreate(BaseModel):
    name: str
    age: int
    email: Optional[str] = None
# 模拟数据库
users_db = {}
@app.get("/")
async def root():
    return {"message": "API 服务运行中"}
# 创建用户
@app.post("/api/users", response_model=User)
async def create_user(user: UserCreate):
    user_id = len(users_db) + 1
    new_user = User(id=user_id, **user.dict())
    users_db[user_id] = new_user
    return new_user
# 获取所有用户
@app.get("/api/users", response_model=List[User])
async def get_users():
    return list(users_db.values())
# 获取单个用户
@app.get("/api/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    return users_db[user_id]
# 更新用户
@app.put("/api/users/{user_id}", response_model=User)
async def update_user(user_id: int, user: UserCreate):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    updated_user = User(id=user_id, **user.dict())
    users_db[user_id] = updated_user
    return updated_user

安装与运行

pip install fastapi uvicorn
uvicorn main:app --reload

Django REST Framework - 全功能框架

创建项目

pip install django djangorestframework
django-admin startproject myapi
cd myapi
python manage.py startapp api

配置settings.py

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',
    'api',
]

创建模型 (api/models.py)

from django.db import models
class User(models.Model):
    name = models.CharField(max_length=100)
    age = models.IntegerField()
    email = models.EmailField(blank=True, null=True)
    created_at = models.DateTimeField(auto_now_add=True)
    def __str__(self):
        return self.name

创建序列化器 (api/serializers.py)

from rest_framework import serializers
from .models import User
class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'name', 'age', 'email', 'created_at']

创建视图 (api/views.py)

from rest_framework import viewsets
from .models import User
from .serializers import UserSerializer
class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

配置路由 (api/urls.py)

from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views
router = DefaultRouter()
router.register(r'users', views.UserViewSet)
urlpatterns = [
    path('api/', include(router.urls)),
]

完整的API服务示例（FastAPI + 数据库）

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from typing import List
# 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# 数据库模型
class UserDB(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    age = Column(Integer)
    email = Column(String)
Base.metadata.create_all(bind=engine)
# Pydantic模型
class User(BaseModel):
    id: int
    name: str
    age: int
    email: str = None
class UserCreate(BaseModel):
    name: str
    age: int
    email: str = None
app = FastAPI()
# 依赖注入
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
# API端点
@app.get("/api/users", response_model=List[User])
async def get_users(db: Session = Depends(get_db)):
    users = db.query(UserDB).all()
    return users
@app.post("/api/users", response_model=User)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = UserDB(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
@app.get("/api/users/{user_id}", response_model=User)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user
@app.put("/api/users/{user_id}", response_model=User)
async def update_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    for key, value in user.dict().items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return db_user
@app.delete("/api/users/{user_id}")
async def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    db.delete(db_user)
    db.commit()
    return {"message": "用户已删除"}

测试API服务

使用curl或Postman测试：

# 获取所有用户
curl http://localhost:8000/api/users
# 创建用户
curl -X POST http://localhost:8000/api/users \
  -H "Content-Type: application/json" \
  -d '{"name":"王五","age":28,"email":"wangwu@example.com"}'
# 获取单个用户
curl http://localhost:8000/api/users/1
# 更新用户
curl -X PUT http://localhost:8000/api/users/1 \
  -H "Content-Type: application/json" \
  -d '{"name":"王五","age":29}'
# 删除用户
curl -X DELETE http://localhost:8000/api/users/1

框架选择建议

框架 适用场景 特点
Flask 小型项目、快速原型 简单灵活，学习成本低
FastAPI 现代API开发、高并发 性能好，自动生成文档，支持异步
Django REST 大型项目、企业应用 功能完整，安全性好，生态完善

部署建议

# 使用Gunicorn部署Flask
pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 app:app
# 使用Uvicorn部署FastAPI
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
# 使用Django部署
python manage.py runserver 0.0.0.0:8000

选择哪个框架取决于你的具体需求,如果是刚开始学习，建议从Flask开始；如果追求性能和现代特性，选择FastAPI；如果是大型企业项目，Django REST Framework是不错的选择。

上一篇Python案例怎么使用Flask框架？

下一篇Python案例怎么校验测试结果？

相关文章

抱歉，评论功能暂时关闭!