コンテンツにスキップ

FastAPIでブログのバックエンドを作る

[質問]

Pythonで、次のようなRestAPIのセットを作ってください。(1)記事を投稿するPOST形式API。(2)投稿した記事のリストを取得するGET形式API。リストの各エントリは、投稿日時と、記事の先頭の何文字かを切り出した文字列で構成されます。(3)個別の記事を取得するGET形式API。(4)個別の記事を削除するAPI。(5)個別の記事を書き換えるPOST形式のAPI。

[回答] Gemini

これまでに実装した機能をすべて統合し、SQLite、キーワード・日付検索、ページネーションを備えた完全なコードです。

このコードを main.py という名前で保存し、pip install fastapi uvicorn sqlalchemy python-dateutil を実行してから、uvicorn main:app --reload で起動してください。

import re
import uuid
from datetime import datetime
from typing import List, Optional

from fastapi import FastAPI, HTTPException, Depends, Query
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, String, DateTime, Text, or_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from dateutil import parser as date_parser

# --- 1. データベース設定 (SQLAlchemy) ---
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# DBモデル定義
class ArticleModel(Base):
    __tablename__ = "articles"
    id = Column(String, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.now)

# テーブルの作成
Base.metadata.create_all(bind=engine)

# --- 2. Pydanticモデル (リクエスト・レスポンス用) ---
class ArticleRequest(BaseModel):
    title: str = Field(..., example="Pythonの基本")
    content: str = Field(..., example="FastAPIは非常に高速なフレームワークです。")

class ArticleSummary(BaseModel):
    id: str
    created_at: datetime
    summary: str

    class Config:
        from_attributes = True

class ArticleDetail(BaseModel):
    id: str
    title: str
    content: str
    created_at: datetime

    class Config:
        from_attributes = True

# --- 3. 依存性注入 (DBセッション管理) ---
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# --- 4. FastAPI アプリケーション ---
app = FastAPI(title="Simple Blog API")

# (1) 記事を投稿する (POST)
@app.post("/articles", response_model=ArticleDetail, status_code=201)
def create_article(article: ArticleRequest, db: Session = Depends(get_db)):
    db_article = ArticleModel(
        id=str(uuid.uuid4()),
        title=article.title,
        content=article.content
    )
    db.add(db_article)
    db.commit()
    db.refresh(db_article)
    return db_article

# (2) 記事のリストを取得する (GET) + 検索 + ページネーション
@app.get("/articles", response_model=List[ArticleSummary])
def list_articles(
    keyword: Optional[str] = Query(None, description="検索ワード。from:YYYY-MM-DD や to:YYYY-MM-DD を含めることができます"),
    page: int = Query(1, ge=1),
    limit: int = Query(10, ge=1, le=100),
    db: Session = Depends(get_db)
):
    query = db.query(ArticleModel)

    if keyword:
        # 日付フィルタの抽出 (from:YYYY-MM-DD / to:YYYY-MM-DD)
        from_match = re.search(r'from:(\d{4}-\d{1,2}-\d{1,2})', keyword)
        to_match = re.search(r'to:(\d{4}-\d{1,2}-\d{1,2})', keyword)

        clean_keyword = keyword

        if from_match:
            clean_keyword = clean_keyword.replace(from_match.group(0), "")
            try:
                start_dt = date_parser.parse(from_match.group(1))
                query = query.filter(ArticleModel.created_at >= start_dt)
            except ValueError:
                pass

        if to_match:
            clean_keyword = clean_keyword.replace(to_match.group(0), "")
            try:
                # 終了日の23:59:59までを対象にする
                end_dt = date_parser.parse(to_match.group(1)).replace(hour=23, minute=59, second=59)
                query = query.filter(ArticleModel.created_at <= end_dt)
            except ValueError:
                pass

        # 残った文字列でタイトルと本文を部分一致検索
        search_text = clean_keyword.strip()
        if search_text:
            query = query.filter(
                or_(
                    ArticleModel.title.contains(search_text),
                    ArticleModel.content.contains(search_text)
                )
            )

    # 並び替えとページネーション適用
    offset = (page - 1) * limit
    articles = query.order_by(ArticleModel.created_at.desc()).offset(offset).limit(limit).all()

    # サマリー(先頭20文字)の作成
    results = []
    for art in articles:
        results.append({
            "id": art.id,
            "created_at": art.created_at,
            "summary": art.content[:20] + "..." if len(art.content) > 20 else art.content
        })
    return results

# (3) 個別の記事を取得する (GET)
@app.get("/articles/{article_id}", response_model=ArticleDetail)
def get_article(article_id: str, db: Session = Depends(get_db)):
    article = db.query(ArticleModel).filter(ArticleModel.id == article_id).first()
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")
    return article

# (4) 個別の記事を削除する (DELETE)
@app.delete("/articles/{article_id}")
def delete_article(article_id: str, db: Session = Depends(get_db)):
    article = db.query(ArticleModel).filter(ArticleModel.id == article_id).first()
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")
    db.delete(article)
    db.commit()
    return {"message": f"Article {article_id} deleted successfully"}

# (5) 個別の記事を書き換える (POST)
@app.post("/articles/{article_id}", response_model=ArticleDetail)
def update_article(article_id: str, update_data: ArticleRequest, db: Session = Depends(get_db)):
    article = db.query(ArticleModel).filter(ArticleModel.id == article_id).first()
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")

    article.title = update_data.title
    article.content = update_data.content
    db.commit()
    db.refresh(article)
    return article

APIの動作確認ガイド

サーバーを起動した状態で http://127.0.0.1:8000/docs(Swagger UI)を開くと、すべてのエンドポイントをGUIからテストできます。

検索とページネーションの例:

  • 「API」という言葉を含む記事を検索: GET /articles?keyword=API
  • 2026年3月の記事のみ表示: GET /articles?keyword=from:2026-03-01 to:2026-03-31
  • 1ページに2件ずつ表示して3ページ目を見る: GET /articles?limit=2&page=3

これでバックエンドの基本機能がすべて揃いました。次は、ユーザー認証(ログイン機能)を追加して、投稿者本人しか記事を削除・更新できないように制限してみますか?