flask

Flask - 軽量な Python Web フレームワーク

概要

Flask は、マイクロサービス、REST API、柔軟な Web アプリケーションの構築を目的とした、Python のマイクロフレームワークです。最小限のコアと充実した拡張機能のエコシステムを備えており、軽量なアーキテクチャ、迅速な開発、コンポーネントを完全に制御できることが求められるプロジェクトに最適です。

主な機能：

• マイクロフレームワークの理念（最小限のコア、拡張可能）
• API 開発向けの Flask-RESTful
• モジュール化されたアプリケーション構造を実現する Blueprint
• Flask-SQLAlchemy による SQLAlchemy 連携
• Jinja2 テンプレートエンジン
• 自動リロード機能を備えた開発用サーバー
• Werkzeug WSGI ツールキットを基盤として採用
• 豊富な拡張機能のエコシステム（Flask-Login、Flask-JWT、Flask-CORS）
• Gunicorn/uWSGI による本番環境へのデプロイ

インストール：

# 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

基本的なアプリケーションパターン

1. 最小限の Flask アプリ

# 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)

実行：

# 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

2. アプリケーションファクトリーパターン

本番環境とテストでの使用を推奨します：

# 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()

実行：

export FLASK_APP=run.py
flask run

3. リクエストとレスポンスの処理

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

Blueprint - モジュール化されたアプリケーション

Blueprint の構造

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

Blueprint の実装

# 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

API 向け Flask-RESTful

基本的な REST API

# 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

リクエストの検証

Marshmallow スキーマ

# 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))

Pydantic による検証

# 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

SQLAlchemy 連携

Flask-SQLAlchemy を使ったモデル

# 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')

認証

Flask-Login によるセッションベース認証

# 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())

JWT 認証

# 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 ファイルのサポート

# .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}')

pytest を使ったテスト

テスト設定

# 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='[email protected]', 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': '[email protected]',
        'password': 'password123'
    })
    token = response.get_json()['access_token']
    return {'Authorization': f'Bearer {token}'}

API テスト

# 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': '[email protected]',
        'name': 'New User',
        'password': 'password123'
    })

    assert response.status_code == 201
    data = response.get_json()
    assert data['email'] == '[email protected]'
    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

デプロイ

Gunicorn を使った本番環境へのデプロイ

# 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'

Docker によるデプロイ

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

ベストプラクティス

1. アプリケーションファクトリーパターンを使う

# ✅ 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__)

2. モジュール化に Blueprint を使う

# ✅ 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

3. すべての入力を検証する

# ✅ 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!

4. シークレットには環境変数を使う

# ✅ GOOD: Environment variables
SECRET_KEY = os.getenv('SECRET_KEY')

# ❌ BAD: Hardcoded secrets
SECRET_KEY = 'hardcoded-secret-key'

5. 適切なエラー処理を実装する

# ✅ 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

リソース

• 公式ドキュメント：https://flask.palletsprojects.com/
• Flask-RESTful：https://flask-restful.readthedocs.io/
• Flask-SQLAlchemy：https://flask-sqlalchemy.palletsprojects.com/
• Flask-JWT-Extended：https://flask-jwt-extended.readthedocs.io/
• Marshmallow：https://marshmallow.readthedocs.io/
• Pydantic：https://pydantic-docs.helpmanual.io/

関連スキル

Flask を使う際は、次の補完的なスキルもご検討ください：

• pytest：フィクスチャとテストクライアントを使った Flask アプリケーションのテスト
• sqlalchemy：Flask-SQLAlchemy 連携によるデータベース ORM パターン
• fastapi-local-dev：高性能 API 向けの最新の非同期代替手段
• django：管理画面と ORM を標準搭載した多機能フレームワーク

Flask の簡易テストパターン（単独利用向けにインライン化）

# 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': '[email protected]'
    })
    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='[email protected]')
        db.session.add(user)
        db.session.commit()

        found = User.query.filter_by(username='bob').first()
        assert found is not None
        assert found.email == '[email protected]'

# Test error handling
def test_404_error(client):
    response = client.get('/nonexistent')
    assert response.status_code == 404
    assert b'Not Found' in response.data

SQLAlchemy の簡易連携（単独利用向けにインライン化）

# 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

FastAPI と Flask の簡単な比較（単独で利用できるように記載）

Flask より FastAPI を選ぶ場合：

• 高い並行性を実現するために async/await が必要な場合
• API ドキュメント（OpenAPI/Swagger）の自動生成を利用したい場合
• 組み込みのデータ検証（Pydantic）が必要な場合
• モダンなマイクロサービスや GraphQL API を構築する場合
• WebSocket または Server-Sent Events のサポートが必要な場合

Flask を使い続ける場合：

• 従来型のサーバーレンダリング Web アプリを構築する場合
• 既存の大規模な Flask コードベースがある場合
• 成熟したエコシステムと豊富なプラグインが必要な場合
• チームが同期型の Python パターンに慣れている場合
• デプロイを簡素化したい場合（非同期ランタイムの複雑さがない）

移行時の考慮事項：

# 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

[pytest、SQLAlchemy、FastAPI の完全なパターンは、同時にデプロイする場合、それぞれのスキルで利用できます]