FastAPIで完璧なブログを構築：訪問者分析

以前の記事で、ブログに全文検索機能を統合し、素晴らしいコンテンツを見つけやすくしました。

これで、ブログの機能が豊富になり、コンテンツが増えるにつれて、読者にとって最も人気のある記事はどれかという新しい疑問が自然に生じます。

読者の興味を理解することは、より質の高いコンテンツを作成するのに役立ちます。

そのため、このチュートリアルでは、ブログに基本的でありながら非常に重要な機能、訪問者トラッキングを追加します。各記事が読まれた回数を記録し、ページにビュー数を表示します。

Google Analyticsのようなサードパーティサービスの使用を検討できます。しかし、バックエンド駆動のトラッキングシステムを自分で構築することで、より多くのデータを自分たちの手に置き、収集したいデータをカスタマイズできます。

始めましょう。

ステップ1：ページビューのデータモデルを作成する

1. データベーステーブルの作成

このテーブルは、各ビューの時間、対応する投稿、および将来の詳細分析のための訪問者情報（IPアドレスやUser Agentなど）を記録します。以下のSQLステートメントをPostgreSQLデータベースで実行して、pageviewテーブルを作成します。

CREATE TABLE "pageview" (
    "id" UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    "createdAt" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    "postId" UUID REFERENCES "post"("id") ON DELETE CASCADE,
    "ipAddress" VARCHAR(45),
    "userAgent" TEXT
);

注意：ON DELETE CASCADEは、投稿が削除されたときに、関連するすべてのページビューレコードも自動的にクリアされることを保証します。

Leapcellでデータベースが作成されている場合、

グラフィカルインターフェイスを使用してSQLステートメントを簡単に実行できます。ウェブサイトのデータベース管理ページに移動し、上記のステートメントをSQLインターフェイスに貼り付けて実行するだけです。

2. PageViewエンティティの作成

次に、models.pyファイルを開き、PageViewモデルを追加し、双方向の関係を確立するためにPostモデルを更新します。

# models.py
import uuid
from datetime import datetime
from typing import Optional, List
from sqlmodel import Field, SQLModel, Relationship

# ... User クラス ...

class Post(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    title: str
    content: str
    createdAt: datetime = Field(default_factory=datetime.utcnow, nullable=False)
    comments: List["Comment"] = Relationship(back_populates="post")
    
    # PageView との1対多の関係を追加
    page_views: List["PageView"] = Relationship(back_populates="post")

# ... Comment クラス ...

class PageView(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    createdAt: datetime = Field(default_factory=datetime.utcnow, nullable=False)
    ipAddress: Optional[str] = Field(max_length=45, default=None)
    userAgent: Optional[str] = Field(default=None)

    # 外部キーを定義し、Post テーブルにリンク
    postId: uuid.UUID = Field(foreign_key="post.id")

    # 多対1の関係を定義
    post: "Post" = Relationship(back_populates="page_views")

main.pyでcreate_db_and_tables関数を構成したため、SQLModelはモデルの変更を自動的に検出し、アプリケーション起動時にデータベーステーブル構造を更新します。そのため、SQLを手動で実行する必要はありません。

ステップ2：トラッキングサービスの K実装

コードをきれいに保つために、ページビューのトラッキング機能用の新しいサービスファイルを作成します。

ページビューに関連するすべてのロジックを処理するために、プロジェクトのルートディレクトリに新しいファイル tracking_service.py を作成します。

# tracking_service.py
import uuid
from typing import List, Dict
from sqlmodel import Session, select, func
from models import PageView

def record_view(post_id: uuid.UUID, ip_address: str, user_agent: str, session: Session):
    """新しいページビューを記録"""
    new_view = PageView(
        postId=post_id,
        ipAddress=ip_address,
        userAgent=user_agent,
    )
    session.add(new_view)
    session.commit()

def get_count_by_post_id(post_id: uuid.UUID, session: Session) -> int:
    """単一投稿の合計ビュー数を取得"""
    statement = select(func.count(PageView.id)).where(PageView.postId == post_id)
    # .one() または .one_or_none() は、単一のスカラー値を返すクエリに必要です
    count = session.exec(statement).one_or_none()
    return count if count is not None else 0

def get_counts_by_post_ids(post_ids: List[uuid.UUID], session: Session) -> Dict[uuid.UUID, int]:
    """効率化のため、複数の投稿のビュー数を一度に取得"""
    if not post_ids:
        return {}
    
    statement = (
        select(PageView.postId, func.count(PageView.id).label("count"))
        .where(PageView.postId.in_(post_ids))
        .group_by(PageView.postId)
    )
    
    results = session.exec(statement).all()
    
    # 結果を {post_id: count} 形式の辞書に変換
    return {post_id: count for post_id, count in results}

get_counts_by_post_idsメソッドは、SQLModel（SQLAlchemy）のfunc.countとgroup_byを使用して、効率的なGROUP BYクエリを実行します。これは、特にホームページで複数の記事のビュー数を表示する必要がある場合、各投稿に対して個別のcountクエリを実行するよりもはるかに高速です。

ステップ3：記事ページでのビュー記録の統合

次に、訪問者が投稿を表示するたびにtracking_serviceのrecord_viewメソッドを呼び出す必要があります。これに最も適した場所は、routers/posts.pyのget_post_by_idルートです。

routers/posts.pyを開き、新しいサービスをインポートして呼び出します。

# routers/posts.py
# ... 他のインポート
import tracking_service # トラッキングサービスをインポート

# ...

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(
    request: Request,
    post_id: uuid.UUID,
    session: Session = Depends(get_session),
    user: dict | None = Depends(get_user_from_session),
):
    post = session.get(Post, post_id)
    if not post:
        # 投稿が見つからない場合の処理
        return HTMLResponse(status_code=404)
        
    comments = comments_service.get_comments_by_post_id(post_id, session)

    # ビューを記録（ファイア・アンド・フォーゲット）
    client_ip = request.client.host
    user_agent = request.headers.get("user-agent", "")
    tracking_service.record_view(post_id, client_ip, user_agent, session)
    
    # ビュー数を取得
    view_count = tracking_service.get_count_by_post_id(post_id, session)

    # Markdownコンテンツを解析
    post.content = markdown2.markdown(post.content)

    return templates.TemplateResponse(
        "post.html",
        {
            "request": request,
            "post": post,
            "title": post.title,
            "user": user,
            "comments": comments,
            "view_count": view_count, # ビュー数をテンプレートに渡す
        },
    )

ステップ4：フロントエンドでのビュー数の表示

記事詳細ページ

前のステップで、view_countを取得し、post.htmlテンプレートに渡しました。あとはテンプレートに表示するだけです。

templates/post.htmlを開き、投稿のメタ情報エリアにビュー数を追加します。

<article class="post-detail">
  <h1>{{ post.title }}</h1>
  <small>{{ post.createdAt.strftime('%Y-%m-%d') }} | Views: {{ view_count }}</small>
  <div class="post-content">{{ post.content | safe }}</div>
</article>

ブログホームページ

ホームページの投稿リストにもビュー数を表示するには、get_all_postsルートにいくつかの調整を加える必要があります。

routers/posts.pyを更新:

# routers/posts.py
# ...

@router.get("/posts", response_class=HTMLResponse)
def get_all_posts(
    request: Request, 
    session: Session = Depends(get_session),
    user: dict | None = Depends(get_user_from_session)
):
    # 1. すべての投稿を取得
    statement = select(Post).order_by(Post.createdAt.desc())
    posts = session.exec(statement).all()
    
    # 2. すべての投稿のIDを取得
    post_ids = [post.id for post in posts]
    
    # 3. ビュー数をバッチで取得
    view_counts = tracking_service.get_counts_by_post_ids(post_ids, session)
    
    # 4. 各投稿オブジェクトにビュー数をアタッチ
    for post in posts:
        post.view_count = view_counts.get(post.id, 0)

    return templates.TemplateResponse(
        "index.html", 
        {"request": request, 
         "posts": posts, 
         "title": "Home", 
         "user": user
        }
    )

# ...

最後に、templates/index.htmlテンプレートを更新してビュー数を表示します。

<div class="post-list">
  {% for post in posts %}
  <article class="post-item">
    <h2><a href="/posts/{{ post.id }}">{{ post.title }}</a></h2>
    <p>{{ post.content[:150] }}...</p>
    <small>{{ post.createdAt.strftime('%Y-%m-%d') }} | Views: {{ post.view_count }}</small>
  </article>
  {% endfor %}
</div>