PyJWT 和 python-jose 在处理JWT令牌处理的时候的差异和具体使用

importjwtimportdatetime#创建一个JWT
payload ={"user_id": 123,"exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)

}

secret= 'your-secret-key'token= jwt.encode(payload, secret, algorithm='HS256')#解码JWT
decoded = jwt.decode(token, secret, algorithms=['HS256'])print(decoded)

from jose importjwtfrom jose.exceptions importJWTError#创建一个JWT
payload ={"user_id": 123,"exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)

}

secret= 'your-secret-key'token= jwt.encode(payload, secret, algorithm='HS256')#解码JWT
try:

    decoded= jwt.decode(token, secret, algorithms=['HS256'])print(decoded)exceptJWTError as e:print(f"Token is invalid: {e}")

pip install python-jose

from jose importjwt, JWTErrorfrom jose.exceptions importExpiredSignatureError#定义密钥和有效负载
secret = 'your-secret-key'payload={"user_id": 123,"exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)

}#生成 JWT
token = jwt.encode(payload, secret, algorithm='HS256')#解码 JWT 并处理可能的异常
try:

    decoded= jwt.decode(token, secret, algorithms=['HS256'])print(decoded)exceptExpiredSignatureError:print("Token has expired")exceptJWTError:print("Token is invalid")

from fastapi importFastAPI, Depends, HTTPException, status, Requestfrom fastapi.responses importJSONResponsefrom jose importJWTError, jwtfrom datetime importdatetime, timedelta



app=FastAPI()#配置项，通常从配置文件中加载
JWT_SECRET_KEY = 'your_jwt_secret_key'JWT_ISSUER= 'your_issuer'JWT_AUDIENCE= 'your_audience'JWT_EXPIRED_DAYS= 7ALGORITHM= 'HS256'

defgenerate_token(user_info: dict, role_type: str):#获取IP地址
    ip = user_info.get('ip', '')#定义声明
    claims ={'id': user_info['id'],'email': user_info['email'],'name': user_info['name'],'nickname': user_info.get('nickname', ''),'phone_number': user_info.get('mobile_phone', ''),'gender': user_info.get('gender', ''),'full_name': user_info.get('full_name', ''),'company_id': user_info.get('company_id', ''),'company_name': user_info.get('company_name', ''),'dept_id': user_info.get('dept_id', ''),'dept_name': user_info.get('dept_name', ''),'role_type': role_type,'ip': ip,'mac_addr': '',  #无法获得Mac地址
        'channel': ''}#定义token过期时间
    expiration = datetime.utcnow() + timedelta(days=JWT_EXPIRED_DAYS)#创建JWT token
    to_encode ={**claims,'iss': JWT_ISSUER,'aud': JWT_AUDIENCE,'exp': expiration

    }



    token= jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=ALGORITHM)return token

@app.post('/token')

asyncdefget_token(request: Request):#模拟的用户信息
    user_info ={'id': 123,'email': 'user@example.com','name': 'John Doe','nickname': 'Johnny','mobile_phone': '123-456-7890','gender': 'Male','full_name': 'Johnathan Doe','company_id': 'ABC123','company_name': 'ABC Corp','dept_id': 'Dept001','dept_name': 'IT','ip': request.client.host

    }

    role_type= 'Admin'token=generate_token(user_info, role_type)

    

    headers={'access-token': token,'Authorization': f'Bearer {token}'}return JSONResponse(content={'token': token}, headers=headers)if __name__ == "__main__":importuvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

from fastapi importDepends, HTTPException, statusfrom jose importJWTError, jwtfrom typing importOptional



JWT_SECRET_KEY= 'your_jwt_secret_key'ALGORITHM= 'HS256'

def get_current_user(token: str =Depends(oauth2_scheme)):try:

        payload= jwt.decode(token, JWT_SECRET_KEY, algorithms=[ALGORITHM])

        user_id: str= payload.get("id")if user_id isNone:raiseHTTPException(

                status_code=status.HTTP_401_UNAUTHORIZED,

                detail="Could not validate credentials",

                headers={"WWW-Authenticate": "Bearer"},

            )returnpayloadexceptJWTError:raiseHTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="Could not validate credentials",

            headers={"WWW-Authenticate": "Bearer"},

        )

from fastapi importFastAPI, Depends, HTTPException, statusfrom fastapi.security importOAuth2PasswordBearer



app=FastAPI()



oauth2_scheme= OAuth2PasswordBearer(tokenUrl="token")#允许匿名访问的路由
@app.get("/public")defread_public_data():return {"message": "This is a public endpoint"}#需要授权的路由
@app.get("/protected")def read_protected_data(current_user: dict =Depends(get_current_user)):return {"message": f"Hello, {current_user['name']}"}#模拟登录生成 JWT 令牌的路由
@app.post("/token")deflogin():#这里省略了身份验证过程，只是直接生成一个 JWT 令牌
    token = jwt.encode({"id": 1, "name": "John Doe"}, JWT_SECRET_KEY, algorithm=ALGORITHM)return {"access_token": token, "token_type": "bearer"}

    #JWT auth, required
app.add_middleware(

        AuthenticationMiddleware,

        backend=JwtAuthMiddleware(),

        on_error=JwtAuthMiddleware.auth_exception_handler,

    )

to_encode ={"sub": "user_id","aud": "your_audience",  #设置aud声明
    "exp": datetime.utcnow() + timedelta(minutes=30)

}



token= jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)

try:

    payload=jwt.decode(

        token,

        settings.TOKEN_SECRET_KEY,

        algorithms=[settings.TOKEN_ALGORITHM],

        audience="your_audience"  #验证aud声明
)print(f"Decoded payload: {payload}")exceptJWTError as e:print(f"Token decode failed: {e}")

to_encode ={"sub": str(user_id),  #确保 user_id 是字符串
    "aud": "your_audience",  #其他字段
    "exp": datetime.utcnow() + timedelta(minutes=30)

}



token= jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)

user_id = 123  #假设 user_id 是一个整数
to_encode ={"sub": str(user_id),  #将 user_id 转换为字符串
    "aud": "your_audience","exp": datetime.utcnow() + timedelta(minutes=30)

}



token= jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)

from pydantic importBaseModelclassUser(BaseModel):

    username: str

    email: str

    password: str#敏感字段
user= User(username="user1", email="user1@example.com", password="secret")#创建一个不包含敏感字段的字典
user_dict = user.dict(exclude={"password"})print(user_dict)

from sqlalchemy importColumn, Stringfrom sqlalchemy.ext.declarative importdeclarative_base



Base=declarative_base()classUser(Base):__tablename__ = 'users'id= Column(String, primary_key=True)

    username=Column(String)

    email=Column(String)

    password= Column(String)  #敏感字段

    defto_dict(self):#移除敏感字段
        return {c.name: getattr(self, c.name) for c in self.__table__.columns if c.name != 'password'}



user= User(id="1", username="user1", email="user1@example.com", password="secret")print(user.to_dict())

classUser:def __init__(self, username, email, password):

        self.username=username

        self.email=email

        self.password= password  #敏感字段

    defto_dict(self):#移除敏感字段
        return{"username": self.username,"email": self.email

        }



user= User(username="user1", email="user1@example.com", password="secret")print(user.to_dict())