fastapi session 用于检查是否登陆的案例 不使用aiosession-memory
时间: 2024-09-11 21:10:43 浏览: 110
FastAPI 中的 session 主要是用来在用户的多个请求之间保持状态的一种方式。在不使用 `aiosession-memory`(这里指的是使用内存存储 session 的方式)的情况下,可以使用数据库或者其他存储方式来管理 session。下面是一个使用了数据库存储 session 来检查用户登录状态的简单示例:
首先,你需要安装依赖,例如 `pydantic` 用于数据验证,`sqlalchemy` 用于数据库操作,以及 `fastapi`。
然后,你可以设置一个数据库模型来存储用户信息和 session 数据,以及 FastAPI 的应用:
```python
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import Optional
from passlib.context import CryptContext
from datetime import datetime, timedelta
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db" # 这里使用 SQLite 数据库作为示例
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# 用户模型
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
hashed_password = Column(String)
# 用户在数据库中的密码
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# 这里省略了密码加密和验证的相关代码
# 创建数据库表
Base.metadata.create_all(bind=engine)
app = FastAPI()
# 依赖项
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 这里省略了用户认证相关的代码
# 检查登录的依赖
async def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = verify_token(token, SECRET_KEY, ALGORITHM)
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except:
raise credentials_exception
user = get_user(db, username=token_data.username)
if user is None:
raise credentials_exception
return user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
```
在上面的代码中,我们首先定义了数据库模型 `User`,然后创建了一个 FastAPI 应用实例。我们定义了一个 `get_db` 的依赖项,用于获取数据库会话。我们还定义了一个 `get_current_user` 的依赖项,用于检查用户是否通过了登录认证。如果用户已经登录,将返回当前用户的信息。
注意,上面的代码是一个简化的示例,省略了用户认证(如密码哈希验证、token 生成和验证等)的实现细节。在实际应用中,你需要使用安全的密码存储方式,并实现 token 的生成和验证逻辑。这些通常涉及到使用 JWT(JSON Web Tokens)和密码哈希库(如 passlib)。
阅读全文