はじめてのAPIを構築して完全なCRUD操作を実装する

皆さんは、Mediumが数百万ものブログ記事をどうやって管理しているのか、Spotifyが膨大なプレイリストをどう追跡しているのか、不思議に思ったことはありませんか？これらすべてのアプリの背後には、CRUD――作成（Create）、読み込み（Read）、更新（Update）、削除（Delete）というシンプルな概念があります。

考えてみましょう：

• Medium：ライターが投稿を作成し、読者がそれを読み、編集者が更新し、時には投稿が削除されます。
• Spotify：あなたはプレイリストを作成し、曲を閲覧し、お気に入りを更新し、飽きた曲を削除します。

このブログ記事で学ぶこと

Flask（Pythonフレームワーク）を使って初めてのCRUDアプリを構築する方法を教えます。仮想の本の図書館を構築し、本を保存します。以下のことができるようになります——

• 新しい本を図書館に追加する（作成）
• コレクションを閲覧する（読み込み）
• 価格タグを更新する（更新）
• 本を削除する（削除）

アプリに必要なツール：

• Flask — 最小限の設定でCRUDアプリを構築できるPythonフレームワークです。
• データベース — 本のデータを保存するために使用します。このブログ記事ではPostgresデータベースを使用します。
• SQLAlchemy — Flaskアプリがデータベースと通信できるようにします。

はじめてのFlaskアプリ

まずは、たった6行のコードでFlaskアプリを書いてみましょう。

# dummy-app.py

from flask import Flask
app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello! Welcome to my dummy flask app!'

if __name__ == '__main__':
    app.run(debug=True)

分解してみましょう：

1. from flask import Flask - ツールボックスからFlaskを取り出しています。
2. app = Flask(__name__) - ウェブアプリケーションを作成しています。
3. @app.route('/') - 「誰かがhomepage('/')を訪れたら」とFlaskに指示しています。
4. 下の関数がFlaskに何を返すかを教えます。

ルートとは？

ルートとはウェブアドレスのことです。サイト訪問者を特定の場所に連れて行くためのアドレスです。Flaskでは次のようにルートを作成します：

# dummy-app.py
@app.route('/hello')
def say_hello():
    return "Hi there!"
    
@app.route('/bye')
def say_bye():
    return "bye there!"

ダミーアプリを実行してみましょう。アプリを実行してウェブページにアクセスする前に、flaskをインストールする必要があります。

pip install flask

これで簡単に実行できます。

python dummy-app.py

ルートについて理解したところで、ウェブアプリで最も重要なルートを紹介します。

• GET: 情報を閲覧する時
• POST: 新しい情報を作成する時
• PUT/PATCH: 情報を更新する時
• DELETE: 情報を削除する時

このブログ記事では、これらのルートを使用して仮想の本図書館アプリを構築する方法を紹介します。では始めましょう！

プロジェクトのセットアップ

• プロジェクトディレクトリとファイルを作成する

# プロジェクトを作成
mkdir crud_app
cd crud_app
touch app.py requirements.txt

• 仮想環境を作成する

python3 -m venv crud
# 仮想環境を有効化する
source crud/bin/activate

• 以下のコードをapp.pyファイルに貼り付ける

# app.py
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://postgres:password@localhost:5432/postgres'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 本のモデル
class Book(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    author = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'author': self.author,
            'price': self.price,
            'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S')
        }

# データベーステーブルを作成する
with app.app_context():
    db.create_all()
# 基本ルート
@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"

if __name__ == '__main__':
    app.run(debug=True)

このコードで行っていること

• Flaskとデータベース接続の設定。SQLALCHEMY_DATABASE_URIでPostgres接続文字列を指定しています。
• 本の情報をデータベースに保存するためのデータベーススキーマを作成 — class Book(db.Model)
• アプリを開始する前にテーブルを作成 — db.create_all()
• アプリをデバッグモードで開始 — debug=True
• requirements.txtファイルから依存関係をインストールする

# requirements.txt
flask
flask-sqlalchemy
psycopg2-binary

pip install -r requirements.txt

このアプリはPostgresデータベースを必要とするため、アプリを開始する前にそれが利用可能であることを確認する必要があります。このデモでは、PostgresのDockerコンテナを使用します。

Postgresデータベースはユーザー名、パスワード、およびデータベース名を入力として受け取ります

docker run --name flask_postgres -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=password -e POSTGRES_DB=postgres -p 5432:5432 -d postgres

データベースコンテナが実行されているので、アプリを実行できます。

• アプリを実行する。

python app.py

POSTルート

書籍のレコードを作成するには、POSTルートを使用できます。レコードを作成するためのPOSTルートをapp.pyに追加しましょう。

# CREATE - 新しい書籍を追加
@app.route('/books', methods=['POST'])
def create_book():
    try:
        data = request.get_json()
        
        # 必須フィールドのバリデーション
        if not all(key in data for key in ['title', 'author', 'price']):
            return jsonify({'error': '必須フィールドが不足しています'}), 400
        
        new_book = Book(
            title=data['title'],
            author=data['author'],
            price=data['price']
        )
        
        db.session.add(new_book)
        db.session.commit()
        
        return jsonify({
            'message': '書籍が正常に作成されました',
            'book': new_book.to_dict()
        }), 201
        
    except Exception as e:
        return jsonify({'error': str(e)}), 400

• 書籍レコードの作成

curlコマンドを実行するか、Postmanや**EchoAPI** VScode拡張機能のようなアプリを使ってAPIコールを行うことができます。

curl -X POST http://127.0.0.1:5000/books \
-H "Content-Type: application/json" \
-d '{"title": "華麗なるギャツビー", "author": "F. スコット・フィッツジェラルド", "price": 9.99}'

curl -X POST http://127.0.0.1:5000/books \
-H "Content-Type: application/json" \
-d '{ "title": "アラバマ物語", "author": "ハーパー・リー", "price": 11.99 }'

ルート取得

これらの本の記録は、読み取り用ルートを使用してクエリを実行できます。

# 読み取り - すべての本を取得
@app.route('/books', methods=['GET'])
def get_books():
    books = Book.query.all()
    return jsonify([book.to_dict() for book in books])

# 読み取り - 特定の本を取得
@app.route('/books/<int:id>', methods=['GET'])
def get_book(id):
    book = Book.query.get_or_404(id)
    return jsonify(book.to_dict())

• /booksルートにクエリを投げると、すべての本の記録を取得できます。

curl http://127.0.0.1:5000/books

• 個々の本はIDを使用してクエリすることができます

# curl http://127.0.0.1:5000/books/{book_id}
curl http://127.0.0.1:5000/books/1
curl http://127.0.0.1:5000/books/3
curl http://127.0.0.1:5000/books/4

更新ルート

PUT HTTPリクエストを使用して既存の本のレコードを更新する必要がある場合。

# 更新 - 本を更新
@app.route('/books/<int:id>', methods=['PUT'])
def update_book(id):
    book = Book.query.get_or_404(id)
    data = request.get_json()
    
    try:
        if 'title' in data:
            book.title = data['title']
        if 'author' in data:
            book.author = data['author']
        if 'price' in data:
            book.price = data['price']
            
        db.session.commit()
        return jsonify({
            'message': 'Book updated successfully',
            'book': book.to_dict()
        })
        
    except Exception as e:
        return jsonify({'error': str(e)}), 400

「To Kill a Mockingbird」の価格を更新しましょう。

curl -X PUT http://127.0.0.1:5000/books/4 \
-H "Content-Type: application/json" \
-d '{"price": 14.99}'

削除ルート

同様に、本のレコードを削除することができます。

# DELETE - 本を削除する
@app.route('/books/<int:id>', methods=['DELETE'])
def delete_book(id):
    book = Book.query.get_or_404(id)
    
    try:
        db.session.delete(book)
        db.session.commit()
        return jsonify({'message': '本が正常に削除されました'})
    except Exception as e:
        return jsonify({'error': str(e)}), 400

本のレコードを削除するには、次のCURLコマンドを使用できます。

curl -X DELETE http://127.0.0.1:5000/books/3

一括操作

同時に複数の本を作成するために

• app.py に以下のコードを追加します。

# 一括作成 - 複数の本を追加する
@app.route('/books/bulk', methods=['POST'])
def create_multiple_books():
    try:
        data = request.get_json()
        
        if not isinstance(data, list):
            return jsonify({'error': '本のリストが期待されています'}), 400
        
        created_books = []
        
        for book_data in data:
            # 各本のバリデーション
            if not all(key in book_data for key in ['title', 'author', 'price']):
                return jsonify({'error': f'本に必要なフィールドが不足しています: {book_data}'}), 400
            
            new_book = Book(
                title=book_data['title'],
                author=book_data['author'],
                price=book_data['price']
            )
            
            db.session.add(new_book)
            created_books.append(new_book)
        
        db.session.commit()
        
        return jsonify({
            'message': f'{len(created_books)}冊の本が正常に作成されました',
            'books': [book.to_dict() for book in created_books]
        }), 201
        
    except Exception as e:
        db.session.rollback()  # エラー時にロールバック
        return jsonify({'error': str(e)}), 400

• 一度に複数の本のレコードを作成するための一括POSTリクエストを実行します。

curl -X POST http://127.0.0.1:5000/books/bulk \
-H "Content-Type: application/json" \
-d '[
  {
    "title": "The Great Gatsby",
    "author": "F. Scott Fitzgerald",
    "price": 9.99
  },
  {
    "title": "1984",
    "author": "George Orwell",
    "price": 12.99
  },
  {
    "title": "To Kill a Mockingbird",
    "author": "Harper Lee",
    "price": 11.99
  }
]'

📚✨

• 本の一括更新

# 一括更新 - 複数の本を更新する
@app.route('/books/bulk-update', methods=['PUT'])
def update_multiple_books():
    try:
        data = request.get_json()
        
        if not isinstance(data, list):
            return jsonify({'error': '本の更新リストが期待されています'}), 400
        
        updated_books = []
        
        for book_update in data:
            if 'id' not in book_update:
                return jsonify({'error': '各本の更新にはIDが必要です'}), 400
                
            book = Book.query.get(book_update['id'])
            if not book:
                return jsonify({'error': f'IDが見つかりません: {book_update["id"]}'}), 404
            
            # 提供されているフィールドを更新します
            if 'title' in book_update:
                book.title = book_update['title']
            if 'author' in book_update:
                book.author = book_update['author']
            if 'price' in book_update:
                book.price = book_update['price']
                
            updated_books.append(book)
        
        db.session.commit()
        
        return jsonify({
            'message': f'{len(updated_books)}冊の本が正常に更新されました',
            'books': [book.to_dict() for book in updated_books]
        })
        
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 400

curl -X PUT http://127.0.0.1:5000/books/bulk-update \
-H "Content-Type: application/json" \
-d '[
  {
    "id": 5,
    "price": 14.99
  },
  {
    "id": 6,
    "price": 15.99,
    "author": "George Orwell (Updated)"
  }
]'

今回の記事はここまでです。次回は、完全なWebアプリUIの構築方法と、Flaskアプリでceleryを使ったバックグラウンドタスクの実行についてお話しします。お楽しみに！