flask by bobmatnyc/claude-mpm-skills
npx skills add https://github.com/bobmatnyc/claude-mpm-skills --skill flaskFlask 是一个用于 Python Web 开发的微框架,专为构建微服务、REST API 和灵活的 Web 应用程序而设计。其极简的核心和广泛的扩展生态系统使其成为需要轻量级架构、快速开发和完全控制组件的项目的理想选择。
主要特性 :
安装 :
# Basic Flask
pip install flask
# Flask with common extensions
pip install flask flask-restful flask-sqlalchemy flask-login flask-cors
# With database support
pip install flask flask-sqlalchemy psycopg2-binary # PostgreSQL
# Full microservices stack
pip install flask flask-restful marshmallow flask-jwt-extended redis
# app.py
from flask import Flask, jsonify, request
app = Flask(__name__)
@app.route('/')
def hello():
return jsonify({"message": "Hello, World!"})
@app.route('/api/users/<int:user_id>')
def get_user(user_id):
return jsonify({"id": user_id, "name": f"User {user_id}"})
@app.route('/api/users', methods=['POST'])
def create_user():
data = request.get_json()
return jsonify({"id": 123, **data}), 201
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=5000)
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
运行:
# Development server
python app.py
# Or using flask CLI
export FLASK_APP=app.py
export FLASK_ENV=development
flask run
# Custom port
flask run --port 8000 --host 0.0.0.0
推荐用于生产和测试:
# app/__init__.py
from flask import Flask
from app.config import Config
from app.extensions import db, migrate, jwt
def create_app(config_class=Config):
"""应用工厂模式。"""
app = Flask(__name__)
app.config.from_object(config_class)
# 初始化扩展
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
# 注册蓝图
from app.routes import api_bp, auth_bp
app.register_blueprint(api_bp, url_prefix='/api')
app.register_blueprint(auth_bp, url_prefix='/auth')
return app
# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()
# app/config.py
import os
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret'
class DevelopmentConfig(Config):
DEBUG = True
TESTING = False
class ProductionConfig(Config):
DEBUG = False
TESTING = False
# run.py
from app import create_app
app = create_app()
if __name__ == '__main__':
app.run()
运行:
export FLASK_APP=run.py
flask run
from flask import Flask, request, jsonify, make_response, abort
app = Flask(__name__)
@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
# GET 请求
if request.method == 'GET':
# 查询参数
page = request.args.get('page', 1, type=int)
limit = request.args.get('limit', 10, type=int)
return jsonify({
"page": page,
"limit": limit,
"data": [...]
})
# POST 请求
if request.method == 'POST':
# JSON 请求体
data = request.get_json()
# 验证
if not data or 'name' not in data:
abort(400, description="Missing required field: name")
# 带自定义头的响应
response = make_response(jsonify({"id": 1, **data}), 201)
response.headers['X-Custom-Header'] = 'value'
return response
# 错误处理
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "Resource not found"}), 404
@app.errorhandler(400)
def bad_request(error):
return jsonify({"error": str(error.description)}), 400
app/
├── __init__.py # 应用工厂
├── extensions.py # 扩展实例
├── config.py # 配置
├── models/
│ ├── __init__.py
│ ├── user.py
│ └── product.py
├── routes/
│ ├── __init__.py
│ ├── auth.py # 认证路由
│ ├── users.py # 用户管理路由
│ └── products.py # 产品路由
└── services/
├── __init__.py
├── user_service.py
└── auth_service.py
# app/routes/users.py
from flask import Blueprint, jsonify, request
from app.models.user import User
from app.extensions import db
users_bp = Blueprint('users', __name__)
@users_bp.route('/', methods=['GET'])
def list_users():
"""列出所有用户。"""
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
users = User.query.paginate(page=page, per_page=per_page)
return jsonify({
"users": [u.to_dict() for u in users.items],
"total": users.total,
"page": users.page,
"pages": users.pages
})
@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""根据 ID 获取用户。"""
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())
@users_bp.route('/', methods=['POST'])
def create_user():
"""创建新用户。"""
data = request.get_json()
user = User(
email=data['email'],
name=data['name']
)
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
@users_bp.route('/<int:user_id>', methods=['PUT'])
def update_user(user_id):
"""更新用户。"""
user = User.query.get_or_404(user_id)
data = request.get_json()
if 'name' in data:
user.name = data['name']
if 'email' in data:
user.email = data['email']
db.session.commit()
return jsonify(user.to_dict())
@users_bp.route('/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
"""删除用户。"""
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# app/__init__.py
def create_app():
app = Flask(__name__)
# 注册蓝图
from app.routes.users import users_bp
from app.routes.products import products_bp
app.register_blueprint(users_bp, url_prefix='/api/users')
app.register_blueprint(products_bp, url_prefix='/api/products')
return app
# app/api/resources.py
from flask import request
from flask_restful import Resource, Api, reqparse, fields, marshal_with
from app.models.user import User
from app.extensions import db
# 响应序列化
user_fields = {
'id': fields.Integer,
'email': fields.String,
'name': fields.String,
'created_at': fields.DateTime(dt_format='iso8601')
}
class UserListResource(Resource):
"""用户集合端点。"""
@marshal_with(user_fields)
def get(self):
"""列出所有用户。"""
users = User.query.all()
return users
def post(self):
"""创建新用户。"""
parser = reqparse.RequestParser()
parser.add_argument('email', required=True, help='Email is required')
parser.add_argument('name', required=True, help='Name is required')
parser.add_argument('password', required=True, help='Password is required')
args = parser.parse_args()
user = User(email=args['email'], name=args['name'])
user.set_password(args['password'])
db.session.add(user)
db.session.commit()
return {'id': user.id, 'email': user.email, 'name': user.name}, 201
class UserResource(Resource):
"""单个用户端点。"""
@marshal_with(user_fields)
def get(self, user_id):
"""根据 ID 获取用户。"""
user = User.query.get_or_404(user_id)
return user
@marshal_with(user_fields)
def put(self, user_id):
"""更新用户。"""
user = User.query.get_or_404(user_id)
parser = reqparse.RequestParser()
parser.add_argument('name')
parser.add_argument('email')
args = parser.parse_args()
if args['name']:
user.name = args['name']
if args['email']:
user.email = args['email']
db.session.commit()
return user
def delete(self, user_id):
"""删除用户。"""
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# app/__init__.py
from flask_restful import Api
def create_app():
app = Flask(__name__)
api = Api(app, prefix='/api/v1')
# 注册资源
from app.api.resources import UserListResource, UserResource
api.add_resource(UserListResource, '/users')
api.add_resource(UserResource, '/users/<int:user_id>')
return app
# app/schemas/user_schema.py
from marshmallow import Schema, fields, validate, validates, ValidationError
class UserSchema(Schema):
"""用户验证模式。"""
id = fields.Int(dump_only=True)
email = fields.Email(required=True, validate=validate.Length(max=120))
name = fields.Str(required=True, validate=validate.Length(min=2, max=80))
password = fields.Str(
required=True,
load_only=True,
validate=validate.Length(min=8)
)
created_at = fields.DateTime(dump_only=True)
@validates('email')
def validate_email(self, value):
"""检查邮箱唯一性。"""
from app.models.user import User
if User.query.filter_by(email=value).first():
raise ValidationError('Email already registered')
class UserUpdateSchema(Schema):
"""用户更新模式(部分更新)。"""
email = fields.Email(validate=validate.Length(max=120))
name = fields.Str(validate=validate.Length(min=2, max=80))
password = fields.Str(load_only=True, validate=validate.Length(min=8))
# 在路由中使用
from marshmallow import ValidationError
from app.schemas.user_schema import UserSchema, UserUpdateSchema
user_schema = UserSchema()
users_schema = UserSchema(many=True)
user_update_schema = UserUpdateSchema()
@users_bp.route('/', methods=['POST'])
def create_user():
"""带验证的用户创建。"""
try:
# 验证和反序列化
data = user_schema.load(request.get_json())
except ValidationError as err:
return jsonify({"errors": err.messages}), 400
user = User(**data)
db.session.add(user)
db.session.commit()
# 序列化响应
return user_schema.dump(user), 201
@users_bp.route('/', methods=['GET'])
def list_users():
"""带序列化的用户列表。"""
users = User.query.all()
return jsonify(users_schema.dump(users))
# app/schemas/user_schema_pydantic.py
from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional
from datetime import datetime
class UserCreate(BaseModel):
"""用户创建模式。"""
email: EmailStr
name: str = Field(..., min_length=2, max_length=80)
password: str = Field(..., min_length=8)
@validator('password')
def password_complexity(cls, v):
"""验证密码复杂度。"""
if not any(char.isdigit() for char in v):
raise ValueError('Password must contain at least one digit')
if not any(char.isupper() for char in v):
raise ValueError('Password must contain at least one uppercase letter')
return v
class UserUpdate(BaseModel):
"""用户更新模式。"""
email: Optional[EmailStr] = None
name: Optional[str] = Field(None, min_length=2, max_length=80)
class UserResponse(BaseModel):
"""用户响应模式。"""
id: int
email: str
name: str
created_at: datetime
class Config:
orm_mode = True
# 使用
from pydantic import ValidationError
@users_bp.route('/', methods=['POST'])
def create_user():
"""使用 Pydantic 验证创建用户。"""
try:
user_data = UserCreate(**request.get_json())
except ValidationError as e:
return jsonify({"errors": e.errors()}), 400
user = User(
email=user_data.email,
name=user_data.name
)
user.set_password(user_data.password)
db.session.add(user)
db.session.commit()
return UserResponse.from_orm(user).dict(), 201
# app/models/user.py
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from app.extensions import db
class User(db.Model):
"""用户模型。"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
name = db.Column(db.String(80), nullable=False)
password_hash = db.Column(db.String(200), nullable=False)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 关系
posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
def set_password(self, password):
"""哈希并设置密码。"""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""验证密码。"""
return check_password_hash(self.password_hash, password)
def to_dict(self):
"""序列化为字典。"""
return {
'id': self.id,
'email': self.email,
'name': self.name,
'is_active': self.is_active,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat()
}
def __repr__(self):
return f'<User {self.email}>'
# app/models/post.py
class Post(db.Model):
"""文章模型。"""
__tablename__ = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def to_dict(self):
return {
'id': self.id,
'title': self.title,
'content': self.content,
'user_id': self.user_id,
'author': self.author.name,
'created_at': self.created_at.isoformat()
}
# 数据库操作
@users_bp.route('/<int:user_id>/posts', methods=['GET'])
def get_user_posts(user_id):
"""获取用户文章(带分页)。"""
user = User.query.get_or_404(user_id)
page = request.args.get('page', 1, type=int)
posts = user.posts.paginate(page=page, per_page=20)
return jsonify({
'posts': [p.to_dict() for p in posts.items],
'total': posts.total,
'page': posts.page
})
# 初始化迁移
flask db init
# 创建迁移
flask db migrate -m "Create users table"
# 应用迁移
flask db upgrade
# 回滚
flask db downgrade
# migrations/versions/xxx_create_users.py (自动生成)
def upgrade():
op.create_table(
'users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('email', sa.String(length=120), nullable=False),
sa.Column('name', sa.String(length=80), nullable=False),
sa.Column('password_hash', sa.String(length=200), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('email')
)
op.create_index('ix_users_email', 'users', ['email'])
def downgrade():
op.drop_index('ix_users_email', table_name='users')
op.drop_table('users')
# app/extensions.py
from flask_login import LoginManager
login_manager = LoginManager()
# app/__init__.py
def create_app():
app = Flask(__name__)
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
@login_manager.user_loader
def load_user(user_id):
from app.models.user import User
return User.query.get(int(user_id))
return app
# app/models/user.py
from flask_login import UserMixin
class User(UserMixin, db.Model):
# ... 现有字段 ...
def get_id(self):
return str(self.id)
# app/routes/auth.py
from flask import Blueprint, request, jsonify
from flask_login import login_user, logout_user, login_required, current_user
from app.models.user import User
auth_bp = Blueprint('auth', __name__)
@auth_bp.route('/login', methods=['POST'])
def login():
"""用户登录。"""
data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
return jsonify({"error": "Invalid credentials"}), 401
login_user(user, remember=data.get('remember', False))
return jsonify(user.to_dict())
@auth_bp.route('/logout', methods=['POST'])
@login_required
def logout():
"""用户登出。"""
logout_user()
return jsonify({"message": "Logged out successfully"})
@auth_bp.route('/me', methods=['GET'])
@login_required
def get_current_user():
"""获取当前认证用户。"""
return jsonify(current_user.to_dict())
# app/extensions.py
from flask_jwt_extended import JWTManager
jwt = JWTManager()
# app/__init__.py
def create_app():
app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'super-secret-key'
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)
app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
jwt.init_app(app)
return app
# app/routes/auth.py
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
jwt_required,
get_jwt_identity
)
@auth_bp.route('/login', methods=['POST'])
def login():
"""登录并返回 JWT 令牌。"""
data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
return jsonify({"error": "Invalid credentials"}), 401
access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)
return jsonify({
"access_token": access_token,
"refresh_token": refresh_token,
"user": user.to_dict()
})
@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
"""刷新访问令牌。"""
current_user_id = get_jwt_identity()
access_token = create_access_token(identity=current_user_id)
return jsonify({"access_token": access_token})
@auth_bp.route('/me', methods=['GET'])
@jwt_required()
def get_current_user():
"""从 JWT 获取当前用户。"""
current_user_id = get_jwt_identity()
user = User.query.get_or_404(current_user_id)
return jsonify(user.to_dict())
# 受保护路由示例
@users_bp.route('/<int:user_id>', methods=['DELETE'])
@jwt_required()
def delete_user(user_id):
"""删除用户(需认证)。"""
current_user_id = get_jwt_identity()
# 授权检查
if current_user_id != user_id:
return jsonify({"error": "Unauthorized"}), 403
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# app/config.py
import os
from datetime import timedelta
class Config:
"""基础配置。"""
# Flask
SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key-change-in-production')
# SQLAlchemy
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False
# JWT
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'jwt-secret-change-in-production')
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
# CORS
CORS_ORIGINS = os.getenv('CORS_ORIGINS', '*').split(',')
# 分页
ITEMS_PER_PAGE = 20
MAX_ITEMS_PER_PAGE = 100
# 日志
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
LOG_FILE = os.getenv('LOG_FILE', 'app.log')
class DevelopmentConfig(Config):
"""开发环境配置。"""
DEBUG = True
TESTING = False
SQLALCHEMY_ECHO = True
class TestingConfig(Config):
"""测试环境配置。"""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
class ProductionConfig(Config):
"""生产环境配置。"""
DEBUG = False
TESTING = False
# 严格的安全要求
SECRET_KEY = os.getenv('SECRET_KEY')
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')
# 生产数据库(推荐 PostgreSQL)
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')
@classmethod
def init_app(cls, app):
"""生产环境特定的初始化。"""
# 记录到 syslog 或外部服务
import logging
from logging.handlers import SysLogHandler
syslog_handler = SysLogHandler()
syslog_handler.setLevel(logging.WARNING)
app.logger.addHandler(syslog_handler)
# 配置工厂
config_by_name = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
def get_config(env_name='default'):
"""根据环境名称获取配置。"""
return config_by_name.get(env_name, DevelopmentConfig)
# app/__init__.py
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(get_config(config_name))
return app
# .env
FLASK_APP=run.py
FLASK_ENV=development
SECRET_KEY=your-secret-key-here
JWT_SECRET_KEY=your-jwt-secret-here
DATABASE_URL=postgresql://user:password@localhost/dbname
CORS_ORIGINS=http://localhost:3000,http://localhost:8000
# 加载环境变量
from dotenv import load_dotenv
load_dotenv()
# app/__init__.py
import os
from dotenv import load_dotenv
def create_app():
load_dotenv()
config_name = os.getenv('FLASK_ENV', 'development')
app = Flask(__name__)
app.config.from_object(get_config(config_name))
return app
# app/errors/handlers.py
from flask import jsonify
from werkzeug.exceptions import HTTPException
from sqlalchemy.exc import SQLAlchemyError
def register_error_handlers(app):
"""注册全局错误处理器。"""
@app.errorhandler(404)
def not_found_error(error):
return jsonify({
"error": "Not Found",
"message": "The requested resource was not found"
}), 404
@app.errorhandler(400)
def bad_request_error(error):
return jsonify({
"error": "Bad Request",
"message": str(error.description) if hasattr(error, 'description') else "Invalid request"
}), 400
@app.errorhandler(401)
def unauthorized_error(error):
return jsonify({
"error": "Unauthorized",
"message": "Authentication required"
}), 401
@app.errorhandler(403)
def forbidden_error(error):
return jsonify({
"error": "Forbidden",
"message": "You don't have permission to access this resource"
}), 403
@app.errorhandler(500)
def internal_error(error):
app.logger.error(f'Internal server error: {error}')
db.session.rollback()
return jsonify({
"error": "Internal Server Error",
"message": "An unexpected error occurred"
}), 500
@app.errorhandler(HTTPException)
def handle_http_exception(error):
"""处理所有 HTTP 异常。"""
return jsonify({
"error": error.name,
"message": error.description
}), error.code
@app.errorhandler(SQLAlchemyError)
def handle_db_error(error):
"""处理数据库错误。"""
app.logger.error(f'Database error: {error}')
db.session.rollback()
return jsonify({
"error": "Database Error",
"message": "A database error occurred"
}), 500
# app/__init__.py
def create_app():
app = Flask(__name__)
from app.errors.handlers import register_error_handlers
register_error_handlers(app)
return app
# app/logging_config.py
import logging
from logging.handlers import RotatingFileHandler
import os
def configure_logging(app):
"""配置应用日志。"""
if not app.debug and not app.testing:
# 如果日志目录不存在则创建
if not os.path.exists('logs'):
os.mkdir('logs')
# 带轮转的文件处理器
file_handler = RotatingFileHandler(
'logs/flask_app.log',
maxBytes=10240000, # 10MB
backupCount=10
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s '
'[in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Flask application startup')
# app/__init__.py
def create_app():
app = Flask(__name__)
from app.logging_config import configure_logging
configure_logging(app)
return app
# 在路由中使用
@users_bp.route('/', methods=['POST'])
def create_user():
current_app.logger.info(f'Creating new user: {request.get_json()}')
# ... 创建用户 ...
current_app.logger.info(f'User created successfully: {user.id}')
# conftest.py
import pytest
from app import create_app
from app.extensions import db
from app.models.user import User
@pytest.fixture(scope='session')
def app():
"""创建用于测试的应用。"""
app = create_app('testing')
with app.app_context():
db.create_all()
yield app
db.drop_all()
@pytest.fixture
def client(app):
"""Flask 测试客户端。"""
return app.test_client()
@pytest.fixture
def runner(app):
"""Flask CLI 测试运行器。"""
return app.test_cli_runner()
@pytest.fixture
def db_session(app):
"""用于测试的数据库会话。"""
with app.app_context():
db.session.begin_nested()
yield db.session
db.session.rollback()
@pytest.fixture
def sample_user(db_session):
"""创建示例用户。"""
user = User(email='test@example.com', name='Test User')
user.set_password('password123')
db_session.add(user)
db_session.commit()
return user
@pytest.fixture
def auth_headers(client, sample_user):
"""获取 JWT 认证头。"""
response = client.post('/auth/login', json={
'email': 'test@example.com',
'password': 'password123'
})
token = response.get_json()['access_token']
return {'Authorization': f'Bearer {token}'}
# tests/test_users.py
import pytest
from app.models.user import User
def test_create_user(client):
"""测试用户创建端点。"""
response = client.post('/api/users', json={
'email': 'new@example.com',
'name': 'New User',
'password': 'password123'
})
assert response.status_code == 201
data = response.get_json()
assert data['email'] == 'new@example.com'
assert data['name'] == 'New User'
assert 'id' in data
def test_get_user(client, sample_user):
"""测试获取用户端点。"""
response = client.get(f'/api/users/{sample_user.id}')
assert response.status_code == 200
data = response.get_json()
assert data['id'] == sample_user.id
assert data['email'] == sample_user.email
def test_list_users(client, sample_user):
"""测试列出用户端点。"""
response = client.get('/api/users')
assert response.status_code == 200
data = response.get_json()
assert 'users' in data
assert len(data['users']) > 0
def test_update_user(client, sample_user, auth_headers):
"""测试用户更新端点。"""
response = client.put(
f'/api/users/{sample_user.id}',
json={'name': 'Updated Name'},
headers=auth_headers
)
assert response.status_code == 200
data = response.get_json()
assert data['name'] == 'Updated Name'
def test_delete_user(client, sample_user, auth_headers):
"""测试用户删除端点。"""
response = client.delete(
f'/api/users/{sample_user.id}',
headers=auth_headers
)
assert response.status_code == 204
assert User.query.get(sample_user.id) is None
def test_authentication_required(client, sample_user):
"""测试受保护端点需要认证。"""
response = client.delete(f'/api/users/{sample_user.id}')
assert response.status_code == 401
# 安装 Gunicorn
pip install gunicorn
# 使用 Gunicorn 运行
gunicorn --workers 4 --bind 0.0.0.0:8000 "app:create_app()"
# 带环境变量
gunicorn --workers 4 --bind 0.0.0.0:8000 --env FLASK_ENV=production "app:create_app()"
gunicorn.conf.py:
import multiprocessing
# 服务器套接字
bind = '0.0.0.0:8000'
backlog = 2048
# 工作进程
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = 'sync'
worker_connections = 1000
timeout = 30
keepalive = 2
# 日志
accesslog = 'logs/access.log'
errorlog = 'logs/error.log'
loglevel = 'info'
# 进程命名
proc_name = 'flask-app'
# 服务器机制
daemon = False
pidfile = 'gunicorn.pid'
**Dockerfile
Flask is a micro-framework for Python web development, designed for building microservices, REST APIs, and flexible web applications. Its minimalist core and extensive extension ecosystem make it ideal for projects requiring lightweight architecture, rapid development, and full control over components.
Key Features :
Installation :
# Basic Flask
pip install flask
# Flask with common extensions
pip install flask flask-restful flask-sqlalchemy flask-login flask-cors
# With database support
pip install flask flask-sqlalchemy psycopg2-binary # PostgreSQL
# Full microservices stack
pip install flask flask-restful marshmallow flask-jwt-extended redis
# app.py
from flask import Flask, jsonify, request
app = Flask(__name__)
@app.route('/')
def hello():
return jsonify({"message": "Hello, World!"})
@app.route('/api/users/<int:user_id>')
def get_user(user_id):
return jsonify({"id": user_id, "name": f"User {user_id}"})
@app.route('/api/users', methods=['POST'])
def create_user():
data = request.get_json()
return jsonify({"id": 123, **data}), 201
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=5000)
Run:
# Development server
python app.py
# Or using flask CLI
export FLASK_APP=app.py
export FLASK_ENV=development
flask run
# Custom port
flask run --port 8000 --host 0.0.0.0
Recommended for production and testing:
# app/__init__.py
from flask import Flask
from app.config import Config
from app.extensions import db, migrate, jwt
def create_app(config_class=Config):
"""Application factory pattern."""
app = Flask(__name__)
app.config.from_object(config_class)
# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
# Register blueprints
from app.routes import api_bp, auth_bp
app.register_blueprint(api_bp, url_prefix='/api')
app.register_blueprint(auth_bp, url_prefix='/auth')
return app
# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()
# app/config.py
import os
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret'
class DevelopmentConfig(Config):
DEBUG = True
TESTING = False
class ProductionConfig(Config):
DEBUG = False
TESTING = False
# run.py
from app import create_app
app = create_app()
if __name__ == '__main__':
app.run()
Run:
export FLASK_APP=run.py
flask run
from flask import Flask, request, jsonify, make_response, abort
app = Flask(__name__)
@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
# GET request
if request.method == 'GET':
# Query parameters
page = request.args.get('page', 1, type=int)
limit = request.args.get('limit', 10, type=int)
return jsonify({
"page": page,
"limit": limit,
"data": [...]
})
# POST request
if request.method == 'POST':
# JSON body
data = request.get_json()
# Validation
if not data or 'name' not in data:
abort(400, description="Missing required field: name")
# Custom response with headers
response = make_response(jsonify({"id": 1, **data}), 201)
response.headers['X-Custom-Header'] = 'value'
return response
# Error handling
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "Resource not found"}), 404
@app.errorhandler(400)
def bad_request(error):
return jsonify({"error": str(error.description)}), 400
app/
├── __init__.py # Application factory
├── extensions.py # Extension instances
├── config.py # Configuration
├── models/
│ ├── __init__.py
│ ├── user.py
│ └── product.py
├── routes/
│ ├── __init__.py
│ ├── auth.py # Authentication routes
│ ├── users.py # User management routes
│ └── products.py # Product routes
└── services/
├── __init__.py
├── user_service.py
└── auth_service.py
# app/routes/users.py
from flask import Blueprint, jsonify, request
from app.models.user import User
from app.extensions import db
users_bp = Blueprint('users', __name__)
@users_bp.route('/', methods=['GET'])
def list_users():
"""List all users."""
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
users = User.query.paginate(page=page, per_page=per_page)
return jsonify({
"users": [u.to_dict() for u in users.items],
"total": users.total,
"page": users.page,
"pages": users.pages
})
@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""Get user by ID."""
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())
@users_bp.route('/', methods=['POST'])
def create_user():
"""Create new user."""
data = request.get_json()
user = User(
email=data['email'],
name=data['name']
)
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
@users_bp.route('/<int:user_id>', methods=['PUT'])
def update_user(user_id):
"""Update user."""
user = User.query.get_or_404(user_id)
data = request.get_json()
if 'name' in data:
user.name = data['name']
if 'email' in data:
user.email = data['email']
db.session.commit()
return jsonify(user.to_dict())
@users_bp.route('/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
"""Delete user."""
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# app/__init__.py
def create_app():
app = Flask(__name__)
# Register blueprints
from app.routes.users import users_bp
from app.routes.products import products_bp
app.register_blueprint(users_bp, url_prefix='/api/users')
app.register_blueprint(products_bp, url_prefix='/api/products')
return app
# app/api/resources.py
from flask import request
from flask_restful import Resource, Api, reqparse, fields, marshal_with
from app.models.user import User
from app.extensions import db
# Response serialization
user_fields = {
'id': fields.Integer,
'email': fields.String,
'name': fields.String,
'created_at': fields.DateTime(dt_format='iso8601')
}
class UserListResource(Resource):
"""User collection endpoint."""
@marshal_with(user_fields)
def get(self):
"""List all users."""
users = User.query.all()
return users
def post(self):
"""Create new user."""
parser = reqparse.RequestParser()
parser.add_argument('email', required=True, help='Email is required')
parser.add_argument('name', required=True, help='Name is required')
parser.add_argument('password', required=True, help='Password is required')
args = parser.parse_args()
user = User(email=args['email'], name=args['name'])
user.set_password(args['password'])
db.session.add(user)
db.session.commit()
return {'id': user.id, 'email': user.email, 'name': user.name}, 201
class UserResource(Resource):
"""Single user endpoint."""
@marshal_with(user_fields)
def get(self, user_id):
"""Get user by ID."""
user = User.query.get_or_404(user_id)
return user
@marshal_with(user_fields)
def put(self, user_id):
"""Update user."""
user = User.query.get_or_404(user_id)
parser = reqparse.RequestParser()
parser.add_argument('name')
parser.add_argument('email')
args = parser.parse_args()
if args['name']:
user.name = args['name']
if args['email']:
user.email = args['email']
db.session.commit()
return user
def delete(self, user_id):
"""Delete user."""
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# app/__init__.py
from flask_restful import Api
def create_app():
app = Flask(__name__)
api = Api(app, prefix='/api/v1')
# Register resources
from app.api.resources import UserListResource, UserResource
api.add_resource(UserListResource, '/users')
api.add_resource(UserResource, '/users/<int:user_id>')
return app
# app/schemas/user_schema.py
from marshmallow import Schema, fields, validate, validates, ValidationError
class UserSchema(Schema):
"""User validation schema."""
id = fields.Int(dump_only=True)
email = fields.Email(required=True, validate=validate.Length(max=120))
name = fields.Str(required=True, validate=validate.Length(min=2, max=80))
password = fields.Str(
required=True,
load_only=True,
validate=validate.Length(min=8)
)
created_at = fields.DateTime(dump_only=True)
@validates('email')
def validate_email(self, value):
"""Check email uniqueness."""
from app.models.user import User
if User.query.filter_by(email=value).first():
raise ValidationError('Email already registered')
class UserUpdateSchema(Schema):
"""User update schema (partial updates)."""
email = fields.Email(validate=validate.Length(max=120))
name = fields.Str(validate=validate.Length(min=2, max=80))
password = fields.Str(load_only=True, validate=validate.Length(min=8))
# Usage in routes
from marshmallow import ValidationError
from app.schemas.user_schema import UserSchema, UserUpdateSchema
user_schema = UserSchema()
users_schema = UserSchema(many=True)
user_update_schema = UserUpdateSchema()
@users_bp.route('/', methods=['POST'])
def create_user():
"""Create user with validation."""
try:
# Validate and deserialize
data = user_schema.load(request.get_json())
except ValidationError as err:
return jsonify({"errors": err.messages}), 400
user = User(**data)
db.session.add(user)
db.session.commit()
# Serialize response
return user_schema.dump(user), 201
@users_bp.route('/', methods=['GET'])
def list_users():
"""List users with serialization."""
users = User.query.all()
return jsonify(users_schema.dump(users))
# app/schemas/user_schema_pydantic.py
from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional
from datetime import datetime
class UserCreate(BaseModel):
"""User creation schema."""
email: EmailStr
name: str = Field(..., min_length=2, max_length=80)
password: str = Field(..., min_length=8)
@validator('password')
def password_complexity(cls, v):
"""Validate password complexity."""
if not any(char.isdigit() for char in v):
raise ValueError('Password must contain at least one digit')
if not any(char.isupper() for char in v):
raise ValueError('Password must contain at least one uppercase letter')
return v
class UserUpdate(BaseModel):
"""User update schema."""
email: Optional[EmailStr] = None
name: Optional[str] = Field(None, min_length=2, max_length=80)
class UserResponse(BaseModel):
"""User response schema."""
id: int
email: str
name: str
created_at: datetime
class Config:
orm_mode = True
# Usage
from pydantic import ValidationError
@users_bp.route('/', methods=['POST'])
def create_user():
"""Create user with Pydantic validation."""
try:
user_data = UserCreate(**request.get_json())
except ValidationError as e:
return jsonify({"errors": e.errors()}), 400
user = User(
email=user_data.email,
name=user_data.name
)
user.set_password(user_data.password)
db.session.add(user)
db.session.commit()
return UserResponse.from_orm(user).dict(), 201
# app/models/user.py
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from app.extensions import db
class User(db.Model):
"""User model."""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
name = db.Column(db.String(80), nullable=False)
password_hash = db.Column(db.String(200), nullable=False)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
def set_password(self, password):
"""Hash and set password."""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""Verify password."""
return check_password_hash(self.password_hash, password)
def to_dict(self):
"""Serialize to dictionary."""
return {
'id': self.id,
'email': self.email,
'name': self.name,
'is_active': self.is_active,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat()
}
def __repr__(self):
return f'<User {self.email}>'
# app/models/post.py
class Post(db.Model):
"""Post model."""
__tablename__ = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def to_dict(self):
return {
'id': self.id,
'title': self.title,
'content': self.content,
'user_id': self.user_id,
'author': self.author.name,
'created_at': self.created_at.isoformat()
}
# Database operations
@users_bp.route('/<int:user_id>/posts', methods=['GET'])
def get_user_posts(user_id):
"""Get user posts with pagination."""
user = User.query.get_or_404(user_id)
page = request.args.get('page', 1, type=int)
posts = user.posts.paginate(page=page, per_page=20)
return jsonify({
'posts': [p.to_dict() for p in posts.items],
'total': posts.total,
'page': posts.page
})
# Initialize migrations
flask db init
# Create migration
flask db migrate -m "Create users table"
# Apply migration
flask db upgrade
# Rollback
flask db downgrade
# migrations/versions/xxx_create_users.py (auto-generated)
def upgrade():
op.create_table(
'users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('email', sa.String(length=120), nullable=False),
sa.Column('name', sa.String(length=80), nullable=False),
sa.Column('password_hash', sa.String(length=200), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('email')
)
op.create_index('ix_users_email', 'users', ['email'])
def downgrade():
op.drop_index('ix_users_email', table_name='users')
op.drop_table('users')
# app/extensions.py
from flask_login import LoginManager
login_manager = LoginManager()
# app/__init__.py
def create_app():
app = Flask(__name__)
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
@login_manager.user_loader
def load_user(user_id):
from app.models.user import User
return User.query.get(int(user_id))
return app
# app/models/user.py
from flask_login import UserMixin
class User(UserMixin, db.Model):
# ... existing fields ...
def get_id(self):
return str(self.id)
# app/routes/auth.py
from flask import Blueprint, request, jsonify
from flask_login import login_user, logout_user, login_required, current_user
from app.models.user import User
auth_bp = Blueprint('auth', __name__)
@auth_bp.route('/login', methods=['POST'])
def login():
"""User login."""
data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
return jsonify({"error": "Invalid credentials"}), 401
login_user(user, remember=data.get('remember', False))
return jsonify(user.to_dict())
@auth_bp.route('/logout', methods=['POST'])
@login_required
def logout():
"""User logout."""
logout_user()
return jsonify({"message": "Logged out successfully"})
@auth_bp.route('/me', methods=['GET'])
@login_required
def get_current_user():
"""Get current authenticated user."""
return jsonify(current_user.to_dict())
# app/extensions.py
from flask_jwt_extended import JWTManager
jwt = JWTManager()
# app/__init__.py
def create_app():
app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'super-secret-key'
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)
app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
jwt.init_app(app)
return app
# app/routes/auth.py
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
jwt_required,
get_jwt_identity
)
@auth_bp.route('/login', methods=['POST'])
def login():
"""Login and return JWT tokens."""
data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
return jsonify({"error": "Invalid credentials"}), 401
access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)
return jsonify({
"access_token": access_token,
"refresh_token": refresh_token,
"user": user.to_dict()
})
@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
"""Refresh access token."""
current_user_id = get_jwt_identity()
access_token = create_access_token(identity=current_user_id)
return jsonify({"access_token": access_token})
@auth_bp.route('/me', methods=['GET'])
@jwt_required()
def get_current_user():
"""Get current user from JWT."""
current_user_id = get_jwt_identity()
user = User.query.get_or_404(current_user_id)
return jsonify(user.to_dict())
# Protected route example
@users_bp.route('/<int:user_id>', methods=['DELETE'])
@jwt_required()
def delete_user(user_id):
"""Delete user (authenticated)."""
current_user_id = get_jwt_identity()
# Authorization check
if current_user_id != user_id:
return jsonify({"error": "Unauthorized"}), 403
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# app/config.py
import os
from datetime import timedelta
class Config:
"""Base configuration."""
# Flask
SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key-change-in-production')
# SQLAlchemy
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False
# JWT
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'jwt-secret-change-in-production')
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
# CORS
CORS_ORIGINS = os.getenv('CORS_ORIGINS', '*').split(',')
# Pagination
ITEMS_PER_PAGE = 20
MAX_ITEMS_PER_PAGE = 100
# Logging
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
LOG_FILE = os.getenv('LOG_FILE', 'app.log')
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
TESTING = False
SQLALCHEMY_ECHO = True
class TestingConfig(Config):
"""Testing configuration."""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
class ProductionConfig(Config):
"""Production configuration."""
DEBUG = False
TESTING = False
# Strict security requirements
SECRET_KEY = os.getenv('SECRET_KEY')
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')
# Production database (PostgreSQL recommended)
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')
@classmethod
def init_app(cls, app):
"""Production-specific initialization."""
# Log to syslog or external service
import logging
from logging.handlers import SysLogHandler
syslog_handler = SysLogHandler()
syslog_handler.setLevel(logging.WARNING)
app.logger.addHandler(syslog_handler)
# Configuration factory
config_by_name = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
def get_config(env_name='default'):
"""Get configuration by environment name."""
return config_by_name.get(env_name, DevelopmentConfig)
# app/__init__.py
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(get_config(config_name))
return app
# .env
FLASK_APP=run.py
FLASK_ENV=development
SECRET_KEY=your-secret-key-here
JWT_SECRET_KEY=your-jwt-secret-here
DATABASE_URL=postgresql://user:password@localhost/dbname
CORS_ORIGINS=http://localhost:3000,http://localhost:8000
# Load environment variables
from dotenv import load_dotenv
load_dotenv()
# app/__init__.py
import os
from dotenv import load_dotenv
def create_app():
load_dotenv()
config_name = os.getenv('FLASK_ENV', 'development')
app = Flask(__name__)
app.config.from_object(get_config(config_name))
return app
# app/errors/handlers.py
from flask import jsonify
from werkzeug.exceptions import HTTPException
from sqlalchemy.exc import SQLAlchemyError
def register_error_handlers(app):
"""Register global error handlers."""
@app.errorhandler(404)
def not_found_error(error):
return jsonify({
"error": "Not Found",
"message": "The requested resource was not found"
}), 404
@app.errorhandler(400)
def bad_request_error(error):
return jsonify({
"error": "Bad Request",
"message": str(error.description) if hasattr(error, 'description') else "Invalid request"
}), 400
@app.errorhandler(401)
def unauthorized_error(error):
return jsonify({
"error": "Unauthorized",
"message": "Authentication required"
}), 401
@app.errorhandler(403)
def forbidden_error(error):
return jsonify({
"error": "Forbidden",
"message": "You don't have permission to access this resource"
}), 403
@app.errorhandler(500)
def internal_error(error):
app.logger.error(f'Internal server error: {error}')
db.session.rollback()
return jsonify({
"error": "Internal Server Error",
"message": "An unexpected error occurred"
}), 500
@app.errorhandler(HTTPException)
def handle_http_exception(error):
"""Handle all HTTP exceptions."""
return jsonify({
"error": error.name,
"message": error.description
}), error.code
@app.errorhandler(SQLAlchemyError)
def handle_db_error(error):
"""Handle database errors."""
app.logger.error(f'Database error: {error}')
db.session.rollback()
return jsonify({
"error": "Database Error",
"message": "A database error occurred"
}), 500
# app/__init__.py
def create_app():
app = Flask(__name__)
from app.errors.handlers import register_error_handlers
register_error_handlers(app)
return app
# app/logging_config.py
import logging
from logging.handlers import RotatingFileHandler
import os
def configure_logging(app):
"""Configure application logging."""
if not app.debug and not app.testing:
# Create logs directory if it doesn't exist
if not os.path.exists('logs'):
os.mkdir('logs')
# File handler with rotation
file_handler = RotatingFileHandler(
'logs/flask_app.log',
maxBytes=10240000, # 10MB
backupCount=10
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s '
'[in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Flask application startup')
# app/__init__.py
def create_app():
app = Flask(__name__)
from app.logging_config import configure_logging
configure_logging(app)
return app
# Usage in routes
@users_bp.route('/', methods=['POST'])
def create_user():
current_app.logger.info(f'Creating new user: {request.get_json()}')
# ... create user ...
current_app.logger.info(f'User created successfully: {user.id}')
# conftest.py
import pytest
from app import create_app
from app.extensions import db
from app.models.user import User
@pytest.fixture(scope='session')
def app():
"""Create application for testing."""
app = create_app('testing')
with app.app_context():
db.create_all()
yield app
db.drop_all()
@pytest.fixture
def client(app):
"""Flask test client."""
return app.test_client()
@pytest.fixture
def runner(app):
"""Flask CLI test runner."""
return app.test_cli_runner()
@pytest.fixture
def db_session(app):
"""Database session for testing."""
with app.app_context():
db.session.begin_nested()
yield db.session
db.session.rollback()
@pytest.fixture
def sample_user(db_session):
"""Create sample user."""
user = User(email='test@example.com', name='Test User')
user.set_password('password123')
db_session.add(user)
db_session.commit()
return user
@pytest.fixture
def auth_headers(client, sample_user):
"""Get JWT authentication headers."""
response = client.post('/auth/login', json={
'email': 'test@example.com',
'password': 'password123'
})
token = response.get_json()['access_token']
return {'Authorization': f'Bearer {token}'}
# tests/test_users.py
import pytest
from app.models.user import User
def test_create_user(client):
"""Test user creation endpoint."""
response = client.post('/api/users', json={
'email': 'new@example.com',
'name': 'New User',
'password': 'password123'
})
assert response.status_code == 201
data = response.get_json()
assert data['email'] == 'new@example.com'
assert data['name'] == 'New User'
assert 'id' in data
def test_get_user(client, sample_user):
"""Test get user endpoint."""
response = client.get(f'/api/users/{sample_user.id}')
assert response.status_code == 200
data = response.get_json()
assert data['id'] == sample_user.id
assert data['email'] == sample_user.email
def test_list_users(client, sample_user):
"""Test list users endpoint."""
response = client.get('/api/users')
assert response.status_code == 200
data = response.get_json()
assert 'users' in data
assert len(data['users']) > 0
def test_update_user(client, sample_user, auth_headers):
"""Test user update endpoint."""
response = client.put(
f'/api/users/{sample_user.id}',
json={'name': 'Updated Name'},
headers=auth_headers
)
assert response.status_code == 200
data = response.get_json()
assert data['name'] == 'Updated Name'
def test_delete_user(client, sample_user, auth_headers):
"""Test user deletion endpoint."""
response = client.delete(
f'/api/users/{sample_user.id}',
headers=auth_headers
)
assert response.status_code == 204
assert User.query.get(sample_user.id) is None
def test_authentication_required(client, sample_user):
"""Test that protected endpoints require authentication."""
response = client.delete(f'/api/users/{sample_user.id}')
assert response.status_code == 401
# Install Gunicorn
pip install gunicorn
# Run with Gunicorn
gunicorn --workers 4 --bind 0.0.0.0:8000 "app:create_app()"
# With environment variable
gunicorn --workers 4 --bind 0.0.0.0:8000 --env FLASK_ENV=production "app:create_app()"
gunicorn.conf.py:
import multiprocessing
# Server socket
bind = '0.0.0.0:8000'
backlog = 2048
# Worker processes
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = 'sync'
worker_connections = 1000
timeout = 30
keepalive = 2
# Logging
accesslog = 'logs/access.log'
errorlog = 'logs/error.log'
loglevel = 'info'
# Process naming
proc_name = 'flask-app'
# Server mechanics
daemon = False
pidfile = 'gunicorn.pid'
Dockerfile:
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application
COPY . .
# Create non-root user
RUN useradd -m -u 1000 flask && chown -R flask:flask /app
USER flask
# Run with Gunicorn
CMD ["gunicorn", "--config", "gunicorn.conf.py", "app:create_app()"]
docker-compose.yml:
version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- FLASK_ENV=production
- DATABASE_URL=postgresql://user:password@db:5432/flask_db
- SECRET_KEY=${SECRET_KEY}
- JWT_SECRET_KEY=${JWT_SECRET_KEY}
depends_on:
- db
- redis
volumes:
- ./logs:/app/logs
db:
image: postgres:15-alpine
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
- POSTGRES_DB=flask_db
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
depends_on:
- app
volumes:
postgres_data:
redis_data:
# app/services/external_api.py
import requests
from flask import current_app
class ExternalAPIService:
"""External API client service."""
def __init__(self, base_url, timeout=10):
self.base_url = base_url
self.timeout = timeout
def get(self, endpoint, **kwargs):
"""GET request to external API."""
url = f"{self.base_url}/{endpoint}"
try:
response = requests.get(url, timeout=self.timeout, **kwargs)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
current_app.logger.error(f'External API error: {e}')
raise
def post(self, endpoint, data, **kwargs):
"""POST request to external API."""
url = f"{self.base_url}/{endpoint}"
try:
response = requests.post(url, json=data, timeout=self.timeout, **kwargs)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
current_app.logger.error(f'External API error: {e}')
raise
# Usage
api_service = ExternalAPIService('https://api.example.com')
@app.route('/proxy/data')
def proxy_data():
"""Proxy request to external service."""
try:
data = api_service.get('data')
return jsonify(data)
except Exception as e:
return jsonify({"error": "External service unavailable"}), 503
# app/routes/health.py
from flask import Blueprint, jsonify
from app.extensions import db
from sqlalchemy import text
health_bp = Blueprint('health', __name__)
@health_bp.route('/health')
def health_check():
"""Basic health check."""
return jsonify({"status": "healthy"}), 200
@health_bp.route('/health/ready')
def readiness_check():
"""Readiness check with dependencies."""
checks = {
"database": False,
"status": "unhealthy"
}
# Check database
try:
db.session.execute(text('SELECT 1'))
checks["database"] = True
except Exception as e:
current_app.logger.error(f'Database health check failed: {e}')
# Overall status
checks["status"] = "healthy" if all([checks["database"]]) else "unhealthy"
status_code = 200 if checks["status"] == "healthy" else 503
return jsonify(checks), status_code
# ✅ GOOD: Application factory
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(get_config(config_name))
return app
# ❌ BAD: Global app instance
app = Flask(__name__)
# ✅ GOOD: Organized blueprints
from app.routes.users import users_bp
from app.routes.products import products_bp
app.register_blueprint(users_bp, url_prefix='/api/users')
app.register_blueprint(products_bp, url_prefix='/api/products')
# ❌ BAD: All routes in single file
# ✅ GOOD: Validation with Marshmallow/Pydantic
try:
data = user_schema.load(request.get_json())
except ValidationError as e:
return jsonify({"errors": e.messages}), 400
# ❌ BAD: No validation
data = request.get_json()
user = User(**data) # Unsafe!
# ✅ GOOD: Environment variables
SECRET_KEY = os.getenv('SECRET_KEY')
# ❌ BAD: Hardcoded secrets
SECRET_KEY = 'hardcoded-secret-key'
# ✅ GOOD: Global error handlers
@app.errorhandler(Exception)
def handle_error(error):
app.logger.error(f'Error: {error}')
return jsonify({"error": "Internal server error"}), 500
# ❌ BAD: Unhandled exceptions
When using Flask, consider these complementary skills:
# Testing Flask with pytest
import pytest
from flask import Flask
from app import create_app, db
@pytest.fixture
def app():
"""Create and configure test app"""
app = create_app('testing')
with app.app_context():
db.create_all()
yield app
db.drop_all()
@pytest.fixture
def client(app):
"""Test client for making requests"""
return app.test_client()
@pytest.fixture
def runner(app):
"""CLI test runner"""
return app.test_cli_runner()
# Test routes
def test_home_page(client):
response = client.get('/')
assert response.status_code == 200
assert b'Welcome' in response.data
def test_api_endpoint(client):
response = client.post('/api/users', json={
'username': 'alice',
'email': 'alice@example.com'
})
assert response.status_code == 201
data = response.get_json()
assert data['username'] == 'alice'
def test_authentication(client):
# Login
response = client.post('/login', data={
'username': 'alice',
'password': 'secret123'
})
assert response.status_code == 302 # Redirect after login
# Access protected route
response = client.get('/dashboard')
assert response.status_code == 200
# Test with database
def test_user_creation(app):
with app.app_context():
user = User(username='bob', email='bob@example.com')
db.session.add(user)
db.session.commit()
found = User.query.filter_by(username='bob').first()
assert found is not None
assert found.email == 'bob@example.com'
# Test error handling
def test_404_error(client):
response = client.get('/nonexistent')
assert response.status_code == 404
assert b'Not Found' in response.data
# Flask-SQLAlchemy setup
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
db = SQLAlchemy(model_class=Base)
def create_app():
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)
with app.app_context():
db.create_all()
return app
# Define models
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)
posts = db.relationship('Post', backref='author', lazy='dynamic')
class Post(db.Model):
__tablename__ = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
# Query patterns
@app.route('/users/<int:user_id>')
def get_user(user_id):
user = db.session.get(User, user_id)
if not user:
abort(404)
return jsonify({
'id': user.id,
'username': user.username,
'email': user.email,
'posts': [{'title': p.title} for p in user.posts]
})
@app.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
user = User(username=data['username'], email=data['email'])
db.session.add(user)
db.session.commit()
return jsonify({'id': user.id}), 201
# Transaction handling
from sqlalchemy.exc import IntegrityError
@app.route('/transfer', methods=['POST'])
def transfer_funds():
try:
# All operations in single transaction
sender = User.query.get_or_404(request.json['sender_id'])
receiver = User.query.get_or_404(request.json['receiver_id'])
amount = request.json['amount']
sender.balance -= amount
receiver.balance += amount
db.session.commit()
return jsonify({'status': 'success'})
except IntegrityError:
db.session.rollback()
return jsonify({'error': 'Transaction failed'}), 400
When to Choose FastAPI over Flask:
When to Stick with Flask:
Migration Considerations:
# Flask pattern
@app.route('/users/<int:user_id>')
def get_user(user_id):
user = User.query.get_or_404(user_id)
return jsonify({'id': user.id, 'name': user.name})
# FastAPI equivalent
@app.get('/users/{user_id}', response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await db.get(User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
[Full pytest, SQLAlchemy, and FastAPI patterns available in respective skills if deployed together]
Weekly Installs
104
Repository
GitHub Stars
20
First Seen
Jan 23, 2026
Security Audits
Gen Agent Trust HubFailSocketPassSnykWarn
Installed on
codex78
opencode77
cursor74
gemini-cli73
github-copilot71
claude-code68
Lark CLI IM 即时消息管理工具:机器人/用户身份操作聊天、消息、文件下载
31,500 周安装