Next.js + FastAPIでAI開発：PrismaとSQLAlchemyのベストプラクティス AIアプリケーション開発において、Next.js（フロントエンド）とFastAPI（バックエンド）の組み合わせが注目されている。PrismaとSQLAlchemyを使ったデータベース設計、API設計、認証・認可、パフォーマンス最適化のベストプラクティスを解説。
要約
AIアプリケーション開発において、フロントエンドにNext.js、バックエンドにFastAPIを使用する構成が広く採用されています。この組み合わせは、フロントエンドの高速なレンダリングと、バックエンドの柔軟なAPI開発を両立させ、特にAI機能を統合する際に大きなメリットを発揮します。本記事では、フロントエンドでPrisma、バックエンドでSQLAlchemyを使用した場合のベストプラクティスなシステム設計、データベース設計、API設計、認証・認可、パフォーマンス最適化のポイントを詳しく解説します。
なぜNext.js + FastAPIなのか？
AIアプリケーション開発でこの組み合わせが選ばれる理由は以下の通りです。
Next.jsの強み
Reactベースのモダンなフレームワーク： コンポーネントベースの開発で保守性が高いSSR/SSG対応： サーバーサイドレンダリングと静的サイト生成でSEOとパフォーマンスを両立App Router： Next.js 13以降の新しいルーティングシステムで、ネストレイアウトやローディング状態管理が簡単TypeScript完全サポート： 型安全性の高い開発が可能
FastAPIの強み
高速なAPI開発： Pydanticによるデータ検証と自動ドキュメント生成非同期処理： async/awaitで高パフォーマンスなAPIを実現Pythonエコシステム： AI/MLライブラリ（OpenAI、LangChain、Transformersなど）との連携が容易自動ドキュメント： Swagger UIとReDocが自動生成されるOpenAPI仕様
システムアーキテクチャ
基本的なシステム構成は以下のようになります。
[
ユーザー
]
<
--
>
[
Next
.
js
Frontend
]
|
|
REST
API
/
WebSocket
|
[
FastAPI
Backend
]
|
+
--
[
SQLAlchemy
ORM
]
|
|
|
v
|
[
PostgreSQL
]
|
+
--
[
AI
Services
]
|
+
--
OpenAI
API
+
--
LangChain
+
--
Vector
DB
(
Pinecone
/
Weaviate
)
データベース設計のベストプラクティス
Prisma vs SQLAlchemy: どちらを使うか
フロントエンド側： Prismaを使わない（Next.jsから直接DBアクセスしない）バックエンド側： SQLAlchemyをメインに使用スキーマ管理： Alembic（SQLAlchemy用）でマイグレーション管理 ただし、以下のようなケースではPrismaを併用することも検討できます：
TypeScriptファーストで型安全性が高い
Prisma StudioでデータをGUIで管理できる
Next.jsのServer ActionsやAPI Routesで直接DBアクセスする場合に便利
PythonのAI/MLライブラリとの連携がスムーズ
複雑なクエリやトランザクション制御が得意
FastAPIとの統合が優れている
SQLAlchemyでのモデル設計例 # backend/models/
user . py
from sqlalchemy import Column , String , DateTime , Boolean
from sqlalchemy . dialects . postgresql import UUID
from datetime import datetime
import uuid
from database import Base
class User ( Base ) :
__tablename__ = "users"
id = Column ( UUID ( as_uuid = True ) , primary_key = True , default = uuid . uuid4 )
email = Column ( String , unique = True , nullable = False , index = True )
hashed_password = Column ( String , nullable = False )
full_name = Column ( String )
is_active = Column ( Boolean , default = True )
created_at = Column ( DateTime , default = datetime . utcnow )
updated_at = Column ( DateTime , default = datetime . utcnow , onupdate = datetime . utcnow )
# リレーション設定例
# conversations = relationship("Conversation", back_populates="user")
# backend/models/
conversation . py
from sqlalchemy import Column , String , DateTime , ForeignKey , Text
from sqlalchemy . dialects . postgresql import UUID , JSONB
from sqlalchemy . orm import relationship
from datetime import datetime
import uuid
from database import Base
class Conversation ( Base ) :
__tablename__ = "conversations"
id = Column ( UUID ( as_uuid = True ) , primary_key = True , default = uuid . uuid4 )
user_id = Column ( UUID ( as_uuid = True ) , ForeignKey ( "
users . id
" ) , nullable = False )
title = Column ( String , nullable = False )
created_at = Column ( DateTime , default = datetime . utcnow )
updated_at = Column ( DateTime , default = datetime . utcnow , onupdate = datetime . utcnow )
# AI用のメタデータをJSONBで保存
metadata = Column ( JSONB , default = { } )
# リレーション
user = relationship ( "User" , back_populates = "conversations" )
messages = relationship ( "Message" , back_populates = "conversation" , cascade = "all, delete-orphan" )
class Message ( Base ) :
__tablename__ = "messages"
id = Column ( UUID ( as_uuid = True ) , primary_key = True , default = uuid . uuid4 )
conversation_id = Column ( UUID ( as_uuid = True ) , ForeignKey ( "
conversations . id
" ) , nullable = False )
role = Column ( String , nullable = False ) # "user" or "assistant"
content = Column ( Text , nullable = False )
created_at = Column ( DateTime , default = datetime . utcnow )
# AI生成情報をJSONBで保存
ai_metadata = Column ( JSONB , default = { } ) # token数、モデル名など
conversation = relationship ( "Conversation" , back_populates = "messages" )
データベース接続管理 # backend/
database . py
from sqlalchemy import create_engine
from sqlalchemy . ext . declarative import declarative_base
from sqlalchemy . orm import sessionmaker
from contextlib import contextmanager
import os
DATABASE_URL = os . getenv ( "DATABASE_URL" )
# 非同期用のエンジン
from sqlalchemy . ext . asyncio import create_async_engine , AsyncSession
engine = create_async_engine (
DATABASE_URL ,
echo = True , # 開発環境ではTrue、本番ではFalse
pool_size = 20 ,
max_overflow = 0 ,
pool_pre_ping = True , # 接続の有効性を確認
)
AsyncSessionLocal = sessionmaker (
engine , class_ = AsyncSession , expire_on_commit = False
)
Base = declarative_base ( )
# 依存性注入用
from fastapi import Depends
from sqlalchemy . ext . asyncio import AsyncSession
async def get_db ( ) - > AsyncSession :
async with AsyncSessionLocal ( ) as session :
try :
yield session
finally :
await session . close ( )
FastAPIでのAPI設計
ベストプラクティスなAPI構造 # backend/
main . py
from fastapi import FastAPI
from fastapi . middleware . cors import CORSMiddleware
import uvicorn
app = FastAPI (
title = "AI Application API" ,
description = "Next.js + FastAPI AI Application" ,
version = "1.0.0"
)
# CORS設定
app . add_middleware (
CORSMiddleware ,
allow_origins = [ "
http : // localhost : 3000
" ] , # Next.jsのデフォルトポート
allow_credentials = True ,
allow_methods = [ "*" ] ,
allow_headers = [ "*" ] ,
)
# ルーターの登録
from routers import auth , conversations , ai
app . include_router ( auth . router , prefix = "/api/auth" , tags = [ "auth" ] )
app . include_router ( conversations . router , prefix = "/api/conversations" , tags = [ "conversations" ] )
app . include_router ( ai . router , prefix = "/api/ai" , tags = [ "ai" ] )
@app . get ( "/" )
def read_root ( ) :
return { "message" : "AI Application API" }
if __name__ == "__main__" :
uvicorn . run
( "main:app" , host = "0.0.0.0" , port = 8000 , reload = True )
Pydanticスキーマの定義 # backend/schemas/
conversation . py
from pydantic import BaseModel , Field
from datetime import datetime
from typing import Optional , List
from uuid import UUID
class MessageBase ( BaseModel ) :
role : str = Field ( . . . , description = "user or assistant" )
content : str
class MessageCreate ( MessageBase ) :
pass
class MessageResponse ( MessageBase ) :
id : UUID
conversation_id : UUID
created_at : datetime
class Config :
from_attributes = True
class ConversationBase ( BaseModel ) :
title : str
class ConversationCreate ( ConversationBase ) :
pass
class ConversationResponse ( ConversationBase ) :
id : UUID
user_id : UUID
created_at : datetime
updated_at : datetime
messages : List [ MessageResponse ] = [ ]
class Config :
from_attributes = True
ルーターの実装 # backend/routers/
conversations . py
from fastapi import APIRouter , Depends , HTTPException
from sqlalchemy . ext . asyncio import AsyncSession
from sqlalchemy import select
from typing import List
from uuid import UUID
from database import get_db
from models . conversation import Conversation , Message
from schemas . conversation import ConversationCreate , ConversationResponse
from auth . dependencies import get_current_user
router = APIRouter ( )
@
router . post
( "/" , response_model = ConversationResponse )
async def create_conversation (
conversation : ConversationCreate ,
db : AsyncSession = Depends ( get_db ) ,
current_user = Depends ( get_current_user )
) :
db_conversation = Conversation (
title = conversation . title ,
user_id = current_
user . id
)
db . add ( db_conversation )
await db . commit ( )
await db . refresh ( db_conversation )
return db_conversation
@router . get ( "/" , response_model = List [ ConversationResponse ] )
async def list_conversations (
db : AsyncSession = Depends ( get_db ) ,
current_user = Depends ( get_current_user )
) :
result = await db . execute (
select ( Conversation )
. where ( Conversation . user_id == current_
user . id
)
. order_by ( Conversation . updated_at . desc ( ) )
)
conversations = result . scalars ( ) . all ( )
return conversations
@router . get ( "/{conversation_id}" , response_model = ConversationResponse )
async def get_conversation (
conversation_id : UUID ,
db : AsyncSession = Depends ( get_db ) ,
current_user = Depends ( get_current_user )
) :
result = await db . execute (
select ( Conversation )
. where (
Conversation . id
== conversation_id )
. where ( Conversation . user_id == current_
user . id
)
)
conversation = result . scalar_one_or_none ( )
if not conversation :
raise HTTPException ( status_code = 404 , detail = "Conversation not found" )
return conversation
AI機能の統合
OpenAI APIとの連携 # backend/routers/
ai . py
from fastapi import APIRouter , Depends , HTTPException
from fastapi . responses import StreamingResponse
from sqlalchemy . ext . asyncio import AsyncSession
import openai
import os
from typing import AsyncGenerator
from database import get_db
from
Next.jsでのフロントエンド実装
APIクライアントの作成 // frontend/lib/api.ts
import axios from 'axios' ;
const api = axios . create ( {
baseURL :
process . env . NEXT
_PUBLIC_API_URL || '
http : / / localhost : 8000
' ,
headers : {
'Content-Type' : 'application/json' ,
} ,
} ) ;
// リクエストインターセプター：認証トークンを追加
api . interceptors . request . use (
( config ) => {
const token = localStorage . getItem ( 'token' ) ;
if ( token ) {
config . headers . Authorization = ` Bearer ${ token } ` ;
}
return config ;
} ,
( error ) => Promise . reject ( error )
) ;
// レスポンスインターセプター：エラーハンドリング
api . interceptors . response . use (
( response ) => response ,
async ( error ) => {
if ( error . response ?. status === 401 ) {
// トークン切れの場合、ログインページにリダイレクト
localStorage . removeItem ( 'token' ) ;
window . location . href = '/login' ;
}
return Promise . reject ( error ) ;
}
) ;
export default api ;
React HooksでのAPI呼び出し // frontend/hooks/useConversations.ts
import { useState , useEffect } from 'react' ;
import api from '@/lib/api' ;
interface Conversation {
id : string ;
title : string ;
created_at : string ;
updated_at : string ;
}
export function useConversations ( ) {
const [ conversations , setConversations ] = useState < Conversation [ ] > ( [ ] ) ;
const [ loading , setLoading ] = useState ( true ) ;
const [ error , setError ] = useState < string | null > ( null ) ;
useEffect ( ( ) => {
fetchConversations ( ) ;
} , [ ] ) ;
const fetchConversations = async ( ) => {
try {
setLoading ( true ) ;
const response = await api . get ( '/api/conversations' ) ;
setConversations (
response . data
) ;
} catch ( err : any ) {
setError ( err . message ) ;
} finally {
setLoading ( false ) ;
}
} ;
const createConversation = async ( title : string ) => {
try {
const response = await
api . post
( '/api/conversations' , { title } ) ;
setConversations ( [
response . data
, ... conversations ] ) ;
return
response . data
;
} catch ( err : any ) {
setError ( err . message ) ;
throw err ;
}
} ;
return {
conversations ,
loading ,
error ,
createConversation ,
refetch : fetchConversations ,
} ;
}
認証・認可の実装
JWTトークンによる認証 # backend/auth/
jwt . py
from datetime import datetime , timedelta
from jose import JWTError , jwt
from passlib . context import CryptContext
import os
SECRET_KEY = os . getenv ( "SECRET_KEY" )
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext ( schemes = [ "bcrypt" ] , deprecated = "auto" )
def create_access_token ( data : dict ) :
to_encode = data . copy ( )
expire = datetime . utcnow ( ) + timedelta ( minutes = ACCESS_TOKEN_EXPIRE_MINUTES )
to_encode . update ( { "exp" : expire } )
encoded_jwt = jwt . encode ( to_encode , SECRET_KEY , algorithm = ALGORITHM )
return encoded_jwt
def verify_token ( token : str ) :
try :
payload = jwt . decode ( token , SECRET_KEY , algorithms = [ ALGORITHM ] )
return payload
except JWTError :
return None
def get_password_hash ( password : str ) :
return pwd_context . hash ( password )
def verify_password ( plain_password : str , hashed_password : str ) :
return pwd_context . verify ( plain_password , hashed_password )
パフォーマンス最適化
1. データベース最適化
インデックスの作成： 頻繁に検索されるカラムにインデックスを追加コネクションプール： SQLAlchemyのpool_sizeを適切に設定N+1問題の解決： joinedload()やselectinload()でEager Loadingを利用
2. キャッシュ戦略 # Redisキャッシュの利用
from redis import asyncio as aioredis
import json
redis = await aioredis . from_url ( "redis : //
localhost
" )
async def get_cached_conversation ( conversation_id : str ) :
cached = await redis . get ( f"conversation: { conversation_id } " )
if cached :
return json . loads ( cached )
return None
async def set_cached_conversation ( conversation_id : str , data : dict ) :
await redis . setex (
f"conversation: { conversation_id } " ,
300 , # 5分間キャッシュ
json . dumps ( data )
)
3. 非同期処理の活用
FastAPIのasync defを積極的に利用
長時間かかる処理はBackgroundTasksやCeleryで非同期実行
まとめ Next.js + FastAPIの組み合わせでAIアプリケーションを開発する際のベストプラクティスを紹介しました。以下のポイントを押さえることで、保守性が高くスケーラブルなシステムを構築できます。
バックエンドはSQLAlchemy、フロントエンドはAPI経由でデータアクセス Pydanticでデータ検証と型安全性を確保 非同期処理で高パフォーマンスを実現 JWTでセキュアな認証・認可を実装 Redisキャッシュとインデックスでパフォーマンス最適化 この構成をベースに、自社のニーズに合わせてカスタマイズし、優れたAIアプリケーションを構築しましょう。
models
.
conversation
import
Conversation
,
Message
from
schemas
.
ai
import
ChatRequest
,
ChatResponse
from
auth
.
dependencies
import
get_current_user
router
=
APIRouter
(
)
openai
.
api_key
=
os
.
getenv
(
"OPENAI_API_KEY"
)
router
.
post
(
"/chat"
)
async
def
chat
(
request
:
ChatRequest
,
db
:
AsyncSession
=
Depends
(
get_db
)
,
current_user
=
Depends
(
get_current_user
)
)
:
# ユーザーメッセージを保存
user_message
=
Message
(
conversation_id
=
request
.
conversation_id
,
role
=
"user"
,
content
=
request
.
message
)
db
.
add
(
user_message
)
await
db
.
commit
(
)
# OpenAI API呼び出し
try
:
response
=
await
openai
.
ChatCompletion
.
acreate
(
model
=
"gpt-4"
,
messages
=
[
{
"role"
:
"system"
,
"content"
:
"You are a helpful assistant."
}
,
{
"role"
:
"user"
,
"content"
:
request
.
message
}
]
,
temperature
=
0.7
,
)
assistant_content
=
response
.
choices
[
0
]
.
message
.
content
# AIメッセージを保存
assistant_message
=
Message
(
conversation_id
=
request
.
conversation_id
,
role
=
"assistant"
,
content
=
assistant_content
,
ai_metadata
=
{
"model"
:
"gpt-4"
,
"tokens"
:
response
.
usage
.
total
}
)
db
.
add
(
assistant_message
)
await
db
.
commit
(
)
return
ChatResponse
(
message
=
assistant_content
)
except
Exception
as
e
:
raise
HTTPException
(
status_code
=
500
,
detail
=
str
(
e
)
)
router
.
post
(
"/chat/stream"
)
async
def
chat_stream
(
request
:
ChatRequest
,
db
:
AsyncSession
=
Depends
(
get_db
)
,
current_user
=
Depends
(
get_current_user
)
)
:
async
def
generate
(
)
-
>
AsyncGenerator
[
str
,
None
]
:
response
=
await
openai
.
ChatCompletion
.
acreate
(
model
=
"gpt-4"
,
messages
=
[
{
"role"
:
"user"
,
"content"
:
request
.
message
}
]
,
stream
=
True
,
)
full_content
=
""
async
for
chunk
in
response
:
if
chunk
.
choices
[
0
]
.
delta
.
get
(
"content"
)
:
content
=
chunk
.
choices
[
0
]
.
delta
.
content
full_content
+=
content
yield
f"data:
{
content
}
\n\n"
# ストリーミング完了後にDBに保存
assistant_message
=
Message
(
conversation_id
=
request
.
conversation_id
,
role
=
"assistant"
,
content
=
full_content
)
db
.
add
(
assistant_message
)
await
db
.
commit
(
)
return
StreamingResponse
(
generate
(
)
,
media_type
=
"text/event-stream"
)
Next.js + FastAPIでAI開発：PrismaとSQLAlchemyのベストプラクティス
