Python主要フレームワーク完全攻略【Web・GUI・機械学習別比較】
Python フレームワークは、開発効率を飛躍的に向上させる重要なツールです。Web開発からGUI、機械学習まで、用途別に最適化されたフレームワークが数多く存在します。本記事では、主要なPythonフレームワークを分野別に分類し、実用的なサンプルコードとともに徹底比較解説します。
フレームワークとは
フレームワークの定義
フレームワークは、アプリケーション開発のための基盤となるソフトウェア構造です。以下の特徴があります：
- 効率化: 定型的なコードを自動生成
- 標準化: 一貫したコード構造
- 機能性: 豊富な組み込み機能
- 保守性: メンテナンスしやすい設計
- 拡張性: プラグインやモジュールで機能追加
フレームワークの分類
| 分野 | 主要フレームワーク | 特徴 |
|---|---|---|
| Web開発 | Flask, Django, FastAPI | HTTP処理、ルーティング |
| GUI開発 | Tkinter, PyQt, Kivy | デスクトップアプリ |
| 機械学習 | TensorFlow, PyTorch, Keras | ニューラルネットワーク |
| テスト | pytest, unittest | 自動テスト |
| データ処理 | pandas, Dask | データ分析・処理 |
Web開発フレームワーク
Flask – 軽量・柔軟なマイクロフレームワーク
from flask import Flask, render_template, request, jsonify
app = Flask(__name__)
@app.route('/')
def home():
return '<h1>Flask アプリケーション</h1>'
@app.route('/user/<name>')
def user_profile(name):
return f'<h1>Welcome, {name}!</h1>'
@app.route('/api/data', methods=['GET', 'POST'])
def api_data():
if request.method == 'POST':
data = request.json
return jsonify({'message': 'データ受信完了', 'data': data})
return jsonify({'users': ['Alice', 'Bob', 'Charlie']})
if __name__ == '__main__':
app.run(debug=True)
Flask の特徴:
- 最小限の機能で軽量
- 高い柔軟性とカスタマイズ性
- 学習コストが低い
- 小〜中規模プロジェクトに適している
Django – 高機能フルスタックフレームワーク
# settings.py
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'blog',
]
# models.py
from django.db import models
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return self.title
# views.py
from django.shortcuts import render
from .models import Post
def post_list(request):
posts = Post.objects.all().order_by('-created_at')
return render(request, 'blog/post_list.html', {'posts': posts})
def post_detail(request, pk):
post = Post.objects.get(pk=pk)
return render(request, 'blog/post_detail.html', {'post': post})
# urls.py
from django.urls import path
from . import views
urlpatterns = [
path('', views.post_list, name='post_list'),
path('post/<int:pk>/', views.post_detail, name='post_detail'),
]
Django の特徴:
- バッテリー同梱（多機能標準装備）
- ORM、認証、管理画面が標準
- 大規模プロジェクトに適している
- セキュリティ機能が充実
FastAPI – 高速・モダンAPIフレームワーク
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List
app = FastAPI(title="FastAPI Example", version="1.0.0")
class User(BaseModel):
id: int
name: str
email: str
class UserCreate(BaseModel):
name: str
email: str
# ダミーデータベース
users_db = [
User(id=1, name="Alice", email="alice@example.com"),
User(id=2, name="Bob", email="bob@example.com")
]
@app.get("/users/", response_model=List[User])
def read_users():
return users_db
@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int):
user = next((u for u in users_db if u.id == user_id), None)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.post("/users/", response_model=User)
def create_user(user: UserCreate):
new_user = User(id=len(users_db) + 1, **user.dict())
users_db.append(new_user)
return new_user
# 実行: uvicorn main:app --reload
FastAPI の特徴:
- 高速パフォーマンス
- 自動API文書生成
- 型安全性（Pydantic）
- 非同期処理対応
その他のWebフレームワーク
Tornado – 非同期・高性能
import tornado.ioloop
import tornado.web
import asyncio
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write('<h1>Tornado アプリケーション</h1>')
class AsyncHandler(tornado.web.RequestHandler):
async def get(self):
await asyncio.sleep(1) # 非同期処理のシミュレーション
self.write({"message": "非同期処理完了"})
app = tornado.web.Application([
(r"/", MainHandler),
(r"/async", AsyncHandler),
])
if __name__ == "__main__":
app.listen(8888)
tornado.ioloop.IOLoop.current().start()
Bottle – 超軽量フレームワーク
from bottle import route, run, template, request
@route('/')
def index():
return '<h1>Bottle アプリケーション</h1>'
@route('/hello/<name>')
def hello(name):
return template('<h1>Hello {{name}}!</h1>', name=name)
@route('/form', method='POST')
def handle_form():
username = request.forms.get('username')
return f'<h1>フォーム送信: {username}</h1>'
run(host='localhost', port=8080, debug=True)
GUI開発フレームワーク
Tkinter – Python標準GUIライブラリ
import tkinter as tk
from tkinter import messagebox, ttk
class TodoApp:
def __init__(self, root):
self.root = root
self.root.title("ToDoアプリ")
self.root.geometry("400x300")
# タスクリスト
self.tasks = []
# GUI構築
self.create_widgets()
def create_widgets(self):
# 入力フレーム
input_frame = ttk.Frame(self.root)
input_frame.pack(pady=10)
self.task_entry = ttk.Entry(input_frame, width=30)
self.task_entry.pack(side=tk.LEFT, padx=5)
add_btn = ttk.Button(input_frame, text="追加", command=self.add_task)
add_btn.pack(side=tk.LEFT)
# タスクリスト
self.task_listbox = tk.Listbox(self.root, height=10)
self.task_listbox.pack(pady=10, padx=20, fill=tk.BOTH, expand=True)
# 削除ボタン
delete_btn = ttk.Button(self.root, text="削除", command=self.delete_task)
delete_btn.pack(pady=5)
def add_task(self):
task = self.task_entry.get()
if task:
self.tasks.append(task)
self.task_listbox.insert(tk.END, task)
self.task_entry.delete(0, tk.END)
def delete_task(self):
selection = self.task_listbox.curselection()
if selection:
index = selection[0]
self.task_listbox.delete(index)
del self.tasks[index]
root = tk.Tk()
app = TodoApp(root)
root.mainloop()
PyQt5/PySide2 – 高機能クロスプラットフォームGUI
from PyQt5.QtWidgets import (QApplication, QMainWindow, QVBoxLayout,
QHBoxLayout, QWidget, QPushButton, QLineEdit,
QListWidget, QLabel)
import sys
class ModernTodoApp(QMainWindow):
def __init__(self):
super().__init__()
self.initUI()
def initUI(self):
self.setWindowTitle('Modern ToDo App')
self.setGeometry(100, 100, 400, 500)
# 中央ウィジェット
central_widget = QWidget()
self.setCentralWidget(central_widget)
# レイアウト
layout = QVBoxLayout()
# タイトル
title = QLabel('タスク管理アプリ')
title.setStyleSheet('font-size: 18px; font-weight: bold; padding: 10px;')
layout.addWidget(title)
# 入力エリア
input_layout = QHBoxLayout()
self.task_input = QLineEdit()
self.task_input.setPlaceholderText('新しいタスクを入力...')
self.add_button = QPushButton('追加')
self.add_button.clicked.connect(self.add_task)
input_layout.addWidget(self.task_input)
input_layout.addWidget(self.add_button)
layout.addLayout(input_layout)
# タスクリスト
self.task_list = QListWidget()
layout.addWidget(self.task_list)
# 削除ボタン
self.delete_button = QPushButton('選択したタスクを削除')
self.delete_button.clicked.connect(self.delete_task)
layout.addWidget(self.delete_button)
central_widget.setLayout(layout)
def add_task(self):
task = self.task_input.text()
if task:
self.task_list.addItem(task)
self.task_input.clear()
def delete_task(self):
current_row = self.task_list.currentRow()
if current_row >= 0:
self.task_list.takeItem(current_row)
app = QApplication(sys.argv)
window = ModernTodoApp()
window.show()
sys.exit(app.exec_())
Kivy – モバイル対応GUI
from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.button import Button
from kivy.uix.label import Label
from kivy.uix.textinput import TextInput
class KivyTodoApp(BoxLayout):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.orientation = 'vertical'
self.padding = 20
self.spacing = 10
# タイトル
title = Label(text='Kivy ToDo App', size_hint_y=None, height=50)
self.add_widget(title)
# 入力エリア
self.text_input = TextInput(hint_text='新しいタスクを入力...',
size_hint_y=None, height=40)
self.add_widget(self.text_input)
# 追加ボタン
add_button = Button(text='タスク追加', size_hint_y=None, height=50)
add_button.bind(on_press=self.add_task)
self.add_widget(add_button)
# タスク表示エリア
self.task_layout = BoxLayout(orientation='vertical')
self.add_widget(self.task_layout)
def add_task(self, instance):
task_text = self.text_input.text
if task_text:
task_button = Button(text=task_text, size_hint_y=None, height=40)
task_button.bind(on_press=self.remove_task)
self.task_layout.add_widget(task_button)
self.text_input.text = ''
def remove_task(self, instance):
self.task_layout.remove_widget(instance)
class TodoApp(App):
def build(self):
return KivyTodoApp()
TodoApp().run()
機械学習フレームワーク
TensorFlow/Keras – 深層学習
import tensorflow as tf
from tensorflow import keras
import numpy as np
# データ準備
(x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()
x_train = x_train.astype('float32') / 255.0
x_test = x_test.astype('float32') / 255.0
# モデル構築
model = keras.Sequential([
keras.layers.Flatten(input_shape=(28, 28)),
keras.layers.Dense(128, activation='relu'),
keras.layers.Dropout(0.2),
keras.layers.Dense(10, activation='softmax')
])
# コンパイル
model.compile(optimizer='adam',
loss='sparse_categorical_crossentropy',
metrics=['accuracy'])
# 訓練
model.fit(x_train, y_train, epochs=5, validation_split=0.1)
# 評価
test_loss, test_accuracy = model.evaluate(x_test, y_test, verbose=0)
print(f'テスト精度: {test_accuracy:.3f}')
PyTorch – 柔軟な深層学習
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
# シンプルなニューラルネットワーク
class SimpleNN(nn.Module):
def __init__(self, input_size, hidden_size, output_size):
super(SimpleNN, self).__init__()
self.fc1 = nn.Linear(input_size, hidden_size)
self.relu = nn.ReLU()
self.fc2 = nn.Linear(hidden_size, output_size)
def forward(self, x):
x = self.fc1(x)
x = self.relu(x)
x = self.fc2(x)
return x
# モデル作成
model = SimpleNN(input_size=784, hidden_size=128, output_size=10)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# ダミーデータでの訓練例
X = torch.randn(1000, 784)
y = torch.randint(0, 10, (1000,))
dataset = TensorDataset(X, y)
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
for epoch in range(5):
for batch_X, batch_y in dataloader:
optimizer.zero_grad()
outputs = model(batch_X)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print(f'Epoch {epoch+1}, Loss: {loss.item():.3f}')
scikit-learn – 従来の機械学習
from sklearn.datasets import load_iris, make_classification
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC
from sklearn.metrics import accuracy_score, classification_report
# データ読み込み
iris = load_iris()
X, y = iris.data, iris.target
# データ分割
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# 複数のモデルで比較
models = {
'Random Forest': RandomForestClassifier(n_estimators=100),
'SVM': SVC(kernel='rbf'),
}
for name, model in models.items():
# 訓練
model.fit(X_train, y_train)
# 予測
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
# クロスバリデーション
cv_scores = cross_val_score(model, X, y, cv=5)
print(f'{name}:')
print(f' テスト精度: {accuracy:.3f}')
print(f' CV平均精度: {cv_scores.mean():.3f} ± {cv_scores.std():.3f}')
print()
テストフレームワーク
pytest – モダンなテストフレームワーク
import pytest
# テスト対象の関数
def add(a, b):
return a + b
def divide(a, b):
if b == 0:
raise ValueError("ゼロ除算エラー")
return a / b
class Calculator:
def multiply(self, a, b):
return a * b
def power(self, base, exp):
return base ** exp
# テストクラス
class TestCalculator:
def setup_method(self):
self.calc = Calculator()
def test_add(self):
assert add(2, 3) == 5
assert add(-1, 1) == 0
def test_divide(self):
assert divide(10, 2) == 5
assert divide(7, 2) == 3.5
def test_divide_by_zero(self):
with pytest.raises(ValueError, match="ゼロ除算エラー"):
divide(10, 0)
def test_multiply(self):
assert self.calc.multiply(3, 4) == 12
assert self.calc.multiply(-2, 5) == -10
@pytest.mark.parametrize("base,exp,expected", [
(2, 3, 8),
(3, 2, 9),
(5, 0, 1),
])
def test_power(self, base, exp, expected):
assert self.calc.power(base, exp) == expected
# フィクスチャの例
@pytest.fixture
def sample_data():
return [1, 2, 3, 4, 5]
def test_with_fixture(sample_data):
assert len(sample_data) == 5
assert sum(sample_data) == 15
unittest – Python標準テストフレームワーク
import unittest
class TestMathOperations(unittest.TestCase):
def setUp(self):
"""各テストメソッド実行前に呼ばれる"""
self.test_data = [1, 2, 3, 4, 5]
def tearDown(self):
"""各テストメソッド実行後に呼ばれる"""
pass
def test_addition(self):
self.assertEqual(2 + 3, 5)
self.assertNotEqual(2 + 3, 6)
def test_string_operations(self):
text = "Hello, World!"
self.assertTrue(text.startswith("Hello"))
self.assertIn("World", text)
self.assertRegex(text, r"Hello.*World")
def test_list_operations(self):
self.assertListEqual(self.test_data, [1, 2, 3, 4, 5])
self.assertGreater(len(self.test_data), 3)
def test_exception_handling(self):
with self.assertRaises(ZeroDivisionError):
result = 10 / 0
with self.assertRaisesRegex(ValueError, "invalid literal"):
int("not_a_number")
@unittest.skip("一時的にスキップ")
def test_skipped(self):
self.fail("このテストはスキップされます")
if __name__ == '__main__':
unittest.main()
非同期フレームワーク
asyncio – 非同期プログラミング基盤
import asyncio
import aiohttp
import time
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()
async def process_data(data):
"""データ処理のシミュレーション"""
await asyncio.sleep(0.1)
return len(data)
async def main():
urls = [
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/1'
]
start_time = time.time()
async with aiohttp.ClientSession() as session:
# URLを並行取得
fetch_tasks = [fetch_url(session, url) for url in urls]
responses = await asyncio.gather(*fetch_tasks)
# レスポンスを並行処理
process_tasks = [process_data(response) for response in responses]
lengths = await asyncio.gather(*process_tasks)
end_time = time.time()
print(f"取得したレスポンス数: {len(lengths)}")
print(f"処理時間: {end_time - start_time:.2f}秒")
print(f"レスポンス長: {lengths}")
# 実行
if __name__ == "__main__":
asyncio.run(main())
Celery – 分散タスクキュー
from celery import Celery
import time
# Celeryアプリケーション作成
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add_numbers(x, y):
"""シンプルな足し算タスク"""
return x + y
@app.task
def heavy_computation(n):
"""重い計算のシミュレーション"""
time.sleep(n) # n秒間の処理をシミュレート
return f"計算完了: {n}秒かかりました"
@app.task
def process_email(email_data):
"""メール送信処理のシミュレーション"""
time.sleep(2) # メール送信処理
return f"メール送信完了: {email_data['to']}"
# タスクの実行例
if __name__ == "__main__":
# 非同期でタスクを実行
result1 = add_numbers.delay(4, 4)
result2 = heavy_computation.delay(3)
result3 = process_email.delay({'to': 'user@example.com', 'subject': 'Test'})
print(f"タスクID: {result1.id}")
print(f"結果: {result1.get()}") # 結果を待機
# Celery workerを起動: celery -A tasks worker --loglevel=info
データ処理フレームワーク
pandas – データ分析
import pandas as pd
import numpy as np
# サンプルデータ作成
data = {
'name': ['Alice', 'Bob', 'Charlie', 'Diana', 'Eve'],
'age': [25, 30, 35, 28, 32],
'department': ['Engineering', 'Sales', 'Engineering', 'Marketing', 'Sales'],
'salary': [70000, 65000, 80000, 72000, 68000]
}
df = pd.DataFrame(data)
# 基本的なデータ分析
print("=== データ概要 ===")
print(df.describe())
print("\n=== 部署別平均給与 ===")
dept_avg = df.groupby('department').agg({
'salary': ['mean', 'count'],
'age': 'mean'
}).round(2)
print(dept_avg)
print("\n=== 条件フィルタ ===")
high_earners = df[df['salary'] > 70000]
print(high_earners[['name', 'salary']])
# データ可視化
import matplotlib.pyplot as plt
plt.figure(figsize=(12, 4))
plt.subplot(1, 2, 1)
df.groupby('department')['salary'].mean().plot(kind='bar')
plt.title('部署別平均給与')
plt.xticks(rotation=45)
plt.subplot(1, 2, 2)
plt.scatter(df['age'], df['salary'])
plt.xlabel('年齢')
plt.ylabel('給与')
plt.title('年齢 vs 給与')
plt.tight_layout()
plt.show()
Dask – 大規模データ処理
import dask.dataframe as dd
import dask.array as da
import numpy as np
# 大きなデータセットのシミュレーション
# 実際の使用では、CSVファイルやParquetファイルから読み込み
# Dask DataFrameの作成
large_df = dd.from_pandas(pd.DataFrame({
'x': np.random.randn(1000000),
'y': np.random.randn(1000000),
'category': np.random.choice(['A', 'B', 'C'], 1000000)
}), npartitions=4)
# 遅延評価による計算定義
result = large_df.groupby('category').x.mean()
# 実際の計算実行
print("カテゴリ別平均:")
print(result.compute())
# Dask Arrayでの並列計算
large_array = da.random.random((1000, 1000), chunks=(100, 100))
mean_value = large_array.mean()
print(f"大きな配列の平均: {mean_value.compute()}")
# 複雑な計算チェーン
complex_result = (large_df
.assign(z=large_df.x + large_df.y)
.groupby('category')
.z.std())
print("カテゴリ別標準偏差:")
print(complex_result.compute())
フレームワーク比較・選択指針
Web開発フレームワーク比較
| フレームワーク | 学習コスト | パフォーマンス | 機能性 | 適用規模 | 特徴 |
|---|---|---|---|---|---|
| Flask | 低 | 中 | 低 | 小〜中 | 軽量・柔軟 |
| Django | 高 | 中 | 高 | 中〜大 | 多機能・安全 |
| FastAPI | 中 | 高 | 中 | 小〜大 | 高速・型安全 |
| Tornado | 高 | 高 | 中 | 中〜大 | 非同期・リアルタイム |
GUI開発フレームワーク比較
| フレームワーク | プラットフォーム | 見た目 | 学習コスト | パフォーマンス |
|---|---|---|---|---|
| Tkinter | クロス | 基本的 | 低 | 中 |
| PyQt5/PySide2 | クロス | モダン | 高 | 高 |
| Kivy | クロス+モバイル | カスタム | 中 | 中 |
機械学習フレームワーム比較
| フレームワーク | 用途 | 学習コスト | 柔軟性 | コミュニティ |
|---|---|---|---|---|
| TensorFlow/Keras | 深層学習 | 中 | 高 | 大 |
| PyTorch | 深層学習・研究 | 中〜高 | 非常に高 | 大 |
| scikit-learn | 従来のML | 低 | 中 | 大 |
プロジェクト別推奨フレームワーク
プロトタイプ・小規模プロジェクト
# 推奨構成: Flask + SQLite + Tkinter
from flask import Flask, render_template
import sqlite3
app = Flask(__name__)
@app.route('/')
def home():
return '<h1>プロトタイプアプリ</h1>'
# 簡単なデータベース操作
def init_db():
conn = sqlite3.connect('app.db')
conn.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)')
conn.close()
if __name__ == '__main__':
init_db()
app.run(debug=True)
中規模Webアプリケーション
# 推奨構成: Django + PostgreSQL + Redis
# settings.py
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'myapp_db',
'USER': 'postgres',
'PASSWORD': 'password',
'HOST': 'localhost',
'PORT': '5432',
}
}
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
}
}
# models.py
from django.db import models
from django.contrib.auth.models import User
class Article(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
author = models.ForeignKey(User, on_delete=models.CASCADE)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['-created_at']
高性能API・マイクロサービス
# 推奨構成: FastAPI + PostgreSQL + Docker
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import uvicorn
# データベース設定
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
app = FastAPI(title="High Performance API", version="2.0.0")
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/items/{item_id}")
async def read_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(Item).filter(Item.id == item_id).first()
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
データ分析・機械学習プロジェクト
# 推奨構成: Jupyter + pandas + scikit-learn + matplotlib
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error
import matplotlib.pyplot as plt
import seaborn as sns
class MLPipeline:
def __init__(self):
self.model = RandomForestRegressor(n_estimators=100, random_state=42)
self.is_trained = False
def load_data(self, file_path):
"""データ読み込み"""
self.df = pd.read_csv(file_path)
return self.df
def preprocess(self):
"""前処理"""
# 欠損値処理
self.df = self.df.fillna(self.df.mean())
# 特徴量とターゲットの分離
self.X = self.df.drop('target', axis=1)
self.y = self.df['target']
# 訓練・テストデータ分割
self.X_train, self.X_test, self.y_train, self.y_test = train_test_split(
self.X, self.y, test_size=0.2, random_state=42
)
def train(self):
"""モデル訓練"""
self.model.fit(self.X_train, self.y_train)
self.is_trained = True
def evaluate(self):
"""モデル評価"""
if not self.is_trained:
raise ValueError("モデルが未訓練です")
y_pred = self.model.predict(self.X_test)
mse = mean_squared_error(self.y_test, y_pred)
# 可視化
plt.figure(figsize=(10, 6))
plt.scatter(self.y_test, y_pred, alpha=0.6)
plt.plot([self.y_test.min(), self.y_test.max()],
[self.y_test.min(), self.y_test.max()], 'r--', lw=2)
plt.xlabel('実際の値')
plt.ylabel('予測値')
plt.title(f'予測精度 (MSE: {mse:.3f})')
plt.show()
return mse
# 使用例
# pipeline = MLPipeline()
# pipeline.load_data('data.csv')
# pipeline.preprocess()
# pipeline.train()
# mse = pipeline.evaluate()
デプロイメント・本番環境
Docker化の例
# Dockerfile for FastAPI
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://postgres:password@db:5432/myapp
depends_on:
- db
- redis
db:
image: postgres:13
environment:
POSTGRES_DB: myapp
POSTGRES_USER: postgres
POSTGRES_PASSWORD: password
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:6-alpine
ports:
- "6379:6379"
volumes:
postgres_data:
CI/CD設定例
# .github/workflows/django.yml
name: Django CI/CD
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:13
env:
POSTGRES_PASSWORD: postgres
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: 3.11
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
- name: Run tests
run: |
python manage.py test
- name: Run linting
run: |
flake8 .
black --check .
パフォーマンス最適化
フレームワーク別最適化技法
Flask最適化
from flask import Flask
from flask_caching import Cache
from flask_compress import Compress
app = Flask(__name__)
# キャッシュ設定
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_URL'] = 'redis://localhost:6379/0'
cache = Cache(app)
# Gzip圧縮
Compress(app)
@app.route('/expensive-data')
@cache.cached(timeout=300) # 5分間キャッシュ
def expensive_data():
# 重い処理
import time
time.sleep(2)
return {'data': 'expensive computation result'}
# データベース接続プール
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
engine = create_engine(
'postgresql://user:pass@localhost/db',
poolclass=QueuePool,
pool_size=20,
max_overflow=30
)
Django最適化
# settings.py での最適化設定
import os
# データベース最適化
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'myapp',
'OPTIONS': {
'MAX_CONNS': 20,
'conn_max_age': 600,
}
}
}
# キャッシュ設定
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
}
}
}
# セッション設定
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'default'
# 静的ファイル最適化
STATICFILES_STORAGE = 'django.contrib.staticfiles.storage.ManifestStaticFilesStorage'
# views.py での最適化
from django.views.decorators.cache import cache_page
from django.db import connection
from django.core.cache import cache
@cache_page(60 * 15) # 15分間キャッシュ
def cached_view(request):
return render(request, 'template.html', context)
# N+1問題の解決
def optimized_query():
# select_related for ForeignKey
posts = Post.objects.select_related('author').all()
# prefetch_related for ManyToMany/reverse ForeignKey
posts = Post.objects.prefetch_related('tags').all()
return posts
セキュリティ対策
Webフレームワークセキュリティ
# Flask セキュリティ設定
from flask import Flask
from flask_talisman import Talisman
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
app = Flask(__name__)
# HTTPS強制・セキュリティヘッダー
Talisman(app, force_https=True)
# レート制限
limiter = Limiter(
app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"]
)
# CSRFトークン
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')
@app.route('/api/data')
@limiter.limit("10 per minute")
def protected_api():
return {'data': 'protected endpoint'}
# Django セキュリティ設定 (settings.py)
SECURE_SSL_REDIRECT = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_BROWSER_XSS_FILTER = True
X_FRAME_OPTIONS = 'DENY'
# CSRF保護
CSRF_COOKIE_SECURE = True
SESSION_COOKIE_SECURE = True
# データベース接続暗号化
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'OPTIONS': {
'sslmode': 'require',
}
}
}
監視・ログ管理
ログ設定例
import logging
import logging.config
# Flask ログ設定
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'default': {
'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s',
}
},
'handlers': {
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': 'app.log',
'maxBytes': 1024*1024*10, # 10MB
'backupCount': 10,
'formatter': 'default',
},
'console': {
'class': 'logging.StreamHandler',
'formatter': 'default',
}
},
'root': {
'level': 'INFO',
'handlers': ['file', 'console']
}
}
logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger(__name__)
# APM監視 (例: Sentry)
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
sentry_sdk.init(
dsn="YOUR_SENTRY_DSN",
integrations=[FlaskIntegration()],
traces_sample_rate=1.0
)
# メトリクス収集 (例: Prometheus)
from prometheus_flask_exporter import PrometheusMetrics
app = Flask(__name__)
metrics = PrometheusMetrics(app)
@app.route('/health')
def health_check():
return {'status': 'healthy', 'timestamp': datetime.now().isoformat()}
まとめ
Python フレームワークの選択は、プロジェクトの要件、チームのスキル、将来の拡張性を総合的に考慮して決定することが重要です。各分野で最適化されたフレームワークを活用することで、開発効率と品質を大幅に向上させることができます。
フレームワーク選択の要点:
- プロジェクト規模: 小規模なら軽量、大規模なら高機能
- 開発期間: 短期なら学習コストの低いもの
- パフォーマンス要件: 高負荷なら最適化されたもの
- チームスキル: 既存知識を活かせるもの
- 将来性: アクティブに開発されているもの
分野別推奨フレームワーク:
- Web開発: Flask（小規模）、Django（大規模）、FastAPI（API）
- GUI開発: Tkinter（シンプル）、PyQt（高機能）、Kivy（モバイル対応）
- 機械学習: scikit-learn（従来ML）、TensorFlow/Keras（深層学習）、PyTorch（研究）
- テスト: pytest（モダン）、unittest（標準）
- 非同期: asyncio（基盤）、Celery（分散処理）
成功のための実践ポイント:
- 適切な仮想環境での依存関係管理
- CI/CDパイプラインの構築
- セキュリティ対策の実装
- パフォーマンス監視とログ管理
- コードの品質保持（テスト・リンター）
本記事のフレームワーク比較とサンプルコードを参考に、あなたのプロジェクトに最適な Python 開発環境を構築してください。継続的な学習と実践により、より効率的で高品質なソフトウェア開発が実現できます。
参考文献
- Flask公式ドキュメント
- Django公式ドキュメント
- FastAPI公式ドキュメント
- PyQt5公式ドキュメント
- TensorFlow公式ドキュメント
- pytest公式ドキュメント
- Python Software Foundation
■プロンプトだけでオリジナルアプリを開発・公開してみた！！
■AI時代の第一歩！「AI駆動開発コース」はじめました！
テックジム東京本校で先行開始。
■テックジム東京本校
「武田塾」のプログラミング版といえば「テックジム」。
講義動画なし、教科書なし。「進捗管理とコーチング」で効率学習。
より早く、より安く、しかも対面型のプログラミングスクールです。
＜短期講習＞5日で5万円の「Pythonミニキャンプ」開催中。
＜月1開催＞放送作家による映像ディレクター養成講座
＜オンライン無料＞ゼロから始めるPython爆速講座
