FastAPIセッション使用手机 – 花酒锄作田
前言
Starlette の SessionMiddleware を利用した FastAPI は、Cookie 署名に基づいたセッション認証を実現するのに非常に便利です。基本原則は、署名後にセッション データが順序付けされてクライアントの Cookie に保存され、サーバーが署名を検証して後続のリクエストでデータを復元することで、ユーザーのステータスを「記憶」するという原則です。
クッキー署名
SessionMiddlewareを使用する前に追加でインストールする必要があります itsdangerousこのライブラリは、データの安全な署名とシリアル化に使用され、送信中にデータが利用できないことを保証します。
注: 署名は読み取りのみ可能で、Cookie 内のペイロードは Base64 コードのみです。
Cookie の最大制限は 4kb です。あまり多くの情報を保存しないでください。
セッションサービス終了旅行单電視時間所问。
python -m pip install itsdangerous
main.py
from http import HTTPStatus
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.sessions import SessionMiddleware
from pydantic import BaseModel, Field
import uvicorn
app = FastAPI()
app.add_middleware(
SessionMiddleware,
secret_key="your-secret-key-here", # 务必替换
max_age=300, # Session 有效期（秒）
same_site="lax", # CSRF 防护
https_only=False, # 生产环境使用 HTTPS 时应设为 True，对应 cookie 的 Secure 标志，表示仅通过 HTTPS 发送
)
mock_db = {
"users": [
{
"username": "zhangsan",
"password": "123456",
},
{
"username": "lisi",
"password": "lisi_pass",
},
],
}
class LoginRequest(BaseModel):
username: str = Field(..., description="用户名")
password: str = Field(..., description="密码")
@app.post("/login")
async def login(request: Request, payload: LoginRequest):
# 模拟登录逻辑
flag = False
for user in mock_db["users"]:
if user["username"] == payload.username and user["password"] == payload.password:
flag = True
break
if not flag:
return JSONResponse(content={"error": {"code": HTTPStatus.UNAUTHORIZED, "description": "登录失败"}}, status_code=HTTPStatus.UNAUTHORIZED)
# 登录成功后，将用户名存储到 session 中
request.session["user"] = payload.username
return JSONResponse(content={"message": "登录成功"}, status_code=HTTPStatus.OK)
async def get_current_user(request: Request):
user_id = request.session.get("user")
if not user_id:
raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="未登录")
return {"user_id": user_id}
@app.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
return JSONResponse(content={"user_id": current_user["user_id"]}, status_code=HTTPStatus.OK)
@app.post("/logout")
async def logout(request: Request, current_user: dict = Depends(get_current_user)):
# 退出登录后，清除 session 中的用户信息
# 对于签名 cookie，cookie 实际上还是存在的，客户端依旧可以用未过期的 cookie 来访问资源
request.session.clear()
return JSONResponse(content={"message": "退出成功"}, status_code=HTTPStatus.OK)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)
カールリクエストテスト用
# 登录
$ curl -v -X POST -H 'Content-Type:application/json' -d '{"username": "zhangsan", "password": "123456"}'
Note: Unnecessary use of -X or --request, POST is already inferred.
* Trying 127.0.0.1:8000...
* Connected to 127.0.0.1 (127.0.0.1) port 8000
* using HTTP/1.x
> POST /login HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/8.14.1
> Accept: */*
> Content-Type:application/json
> Content-Length: 46
>
* upload completely sent off: 46 bytes
< HTTP/1.1 200 OK
< date: Tue, 15 Sep 2026 13:01:34 GMT
< server: uvicorn
< content-length: 26
< content-type: application/json
< vary: Cookie
< set-cookie: session=eyJ1c2VyIjogInpoYW5nc2FuIn0=.aqlBrg.dCAxuZomyvLQiaL-dZP9zYL1dVg; path=/; Max-Age=300; httponly; samesite=lax
<
* Connection #0 to host 127.0.0.1 left intact
{"message":"登录成功"}
保護インターフェイスへのアクセス、不带 cookie会设计设计
$ curl -v
* Trying 127.0.0.1:8000...
* Connected to 127.0.0.1 (127.0.0.1) port 8000
* using HTTP/1.x
> GET /profile HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/8.14.1
> Accept: */*
>
* Request completely sent off
< HTTP/1.1 401 Unauthorized
< date: Tue, 15 Sep 2026 13:01:40 GMT
< server: uvicorn
< content-length: 22
< content-type: application/json
< vary: Cookie
<
* Connection #0 to host 127.0.0.1 left intact
{"detail":"未登录"}
# 带 cookie 访问受保护接口
$ curl -v -b 'session=eyJ1c2VyIjogInpoYW5nc2FuIn0=.aqlBrg.dCAxuZomyvLQiaL-dZP9zYL1dVg; path=/; Max-Age=300; httponly; samesite=lax'
* Trying 127.0.0.1:8000...
* Connected to 127.0.0.1 (127.0.0.1) port 8000
* using HTTP/1.x
> GET /profile HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/8.14.1
> Accept: */*
> Cookie: session=eyJ1c2VyIjogInpoYW5nc2FuIn0=.aqlBrg.dCAxuZomyvLQiaL-dZP9zYL1dVg; path=/; Max-Age=300; httponly; samesite=lax
>
* Request completely sent off
< HTTP/1.1 200 OK
< date: Tue, 15 Sep 2026 13:01:45 GMT
< server: uvicorn
< content-length: 22
< content-type: application/json
< vary: Cookie
<
* Connection #0 to host 127.0.0.1 left intact
{"user_id":"zhangsan"}
ログアウト。ここでは署名 Cookie が使用されているため、サービスを単独で放棄することはできません。実際、クライアントはまだ有効期限が切れていない Cookie を使用してリソースにアクセスできますが、クライアントはそれらを自分で削除する必要があります。
$ curl -v -X POST -b 'session=eyJ1c2VyIjogInpoYW5nc2FuIn0=.aqlBrg.dCAxuZomyvLQiaL-dZP9zYL1dVg; path=/; Max-Age=300; httponly; samesite=lax'
* Trying 127.0.0.1:8000...
* Connected to 127.0.0.1 (127.0.0.1) port 8000
* using HTTP/1.x
> POST /logout HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/8.14.1
> Accept: */*
> Cookie: session=eyJ1c2VyIjogInpoYW5nc2FuIn0=.aqlBrg.dCAxuZomyvLQiaL-dZP9zYL1dVg; path=/; Max-Age=300; httponly; samesite=lax
>
* Request completely sent off
< HTTP/1.1 200 OK
< date: Tue, 15 Sep 2026 13:01:57 GMT
< server: uvicorn
< content-length: 26
< content-type: application/json
< vary: Cookie
< set-cookie: session=null; path=/; expires=Thu, 01 Jan 1970 00:00:00 GMT; httponly; samesite=lax
<
* Connection #0 to host 127.0.0.1 left intact
{"message":"退出成功"}
セッションサービス終了画像
上記の Cookie 署名は、做不过主安全废session、任乐最好服务端Session来看でのみ認証できます。
サービス セッションはメモリに情報を保存し、通常は単一プロセスの展開またはローカル開発テストにのみ使用されます。デモンストレーションの目的で、次のコード例は非常に単純です。 MemorySession のセッションでは、電気効果音が発生します。
運用環境では、通常、データベースはセッション情報の保存に使用され、複数のインスタンス間でデータを共有できます。
ユーザーのログイン状態を複数の端末から同時に追い出す必要がある場合、キーIDとしてユーザー名またはユーザーIDとその他の情報を1セットに設定できます。
import time
from http import HTTPStatus
from typing import Dict
from uuid import uuid4
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from starlette.middleware.sessions import SessionMiddleware
app = FastAPI()
SESSION_TTL = 300
SESSION_SECRET_KEY = "your-secret-key-here" # 务必替换
app.add_middleware(
SessionMiddleware,
secret_key=SESSION_SECRET_KEY, # 务必替换
max_age=SESSION_TTL, # Session 有效期（秒）
same_site="lax", # CSRF 防护
https_only=False, # 生产环境使用 HTTPS 时应设为 True，对应 cookie 的 Secure 标志，表示仅通过 HTTPS 发送
)
mock_db = {
"users": [
{
"username": "zhangsan",
"password": "123456", # 在生产环境中，数据传输和存储都不应当使用明文密码
},
{
"username": "lisi",
"password": "lisi_pass",
},
],
}
class MemorySessions:
session_cache: Dict[
str, tuple[str, int]
] = {} # 这里用类属性当单例用，实际生产环境中应注意避免使用这种方式
def set(self, username: str) -> str:
"""设置 session
Returns:
str: session_id
"""
session_id = str(uuid4())
expire_at = int(time.time()) + SESSION_TTL
self.session_cache[session_id] = (username, expire_at)
return session_id
def get(self, sess_id: str) -> str:
"""获取 session
Returns:
str: username
"""
item = self.session_cache.get(sess_id)
if not item:
return ""
username, expire_at = item
if expire_at < int(time.time()):
del self.session_cache[sess_id]
return ""
# SessionMiddleware 只能滑动设置一个相对过期时间，如果要刷新过期时间，建议另行实现
# self.session_cache[sess_id] = (username, int(time.time()) + SESSION_TTL)
return username
def delete(self, sess_id: str = ""):
"""删除 session"""
self.session_cache.pop(sess_id, None)
def clean(self):
"""清除所有 session，踢出所有用户"""
self.session_cache.clear()
class LoginRequest(BaseModel):
username: str = Field(..., description="用户名")
password: str = Field(..., description="密码")
@app.post("/login")
async def login(request: Request, payload: LoginRequest):
# 模拟登录逻辑
flag = False
for user in mock_db["users"]:
if (
user["username"] == payload.username
and user["password"] == payload.password
):
flag = True
break
if not flag:
return JSONResponse(
content={
"error": {"code": HTTPStatus.UNAUTHORIZED, "description": "登录失败"}
},
status_code=HTTPStatus.UNAUTHORIZED,
)
# 登录成功后，将用户名存储到 session 中
sess = MemorySessions()
session_id = sess.set(payload.username)
request.session["session_id"] = session_id
return JSONResponse(content={"message": "登录成功"}, status_code=HTTPStatus.OK)
async def get_current_user(request: Request) -> dict:
"""获取当前登录用户, 用于依赖注入"""
session_id = request.session.get("session_id")
if not session_id:
raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="未登录")
sess = MemorySessions()
user = sess.get(session_id)
if not user:
request.session.clear() # 清除 session 中的用户信息
raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="未登录")
return {"user": user, "session_id": session_id}
@app.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
return JSONResponse(
content={"user": current_user["user"]}, status_code=HTTPStatus.OK
)
@app.post("/logout")
async def logout(request: Request, current_user: dict = Depends(get_current_user)):
"""退出登录"""
# NOTE: 当用户的 cookie 已经过期时, get_current_user 会抛出异常
# 这里假装用户在 cookie 还未过期时主动登出
request.session.clear()
sess = MemorySessions()
sess.delete(current_user["session_id"])
return JSONResponse(content={"message": "退出成功"}, status_code=HTTPStatus.OK)
# @app.get("/sessions")
# async def get_cached():
# """获取所有 session, 仅用于调试"""
# sess = MemorySessions()
# return JSONResponse(
# content={"sessions": sess.session_cache}, status_code=HTTPStatus.OK
# )
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)
