第一次提交

This commit is contained in:
wangjiming
2026-07-27 09:12:47 +08:00
commit b4ff5db17b
579 changed files with 48768 additions and 0 deletions

View File

@@ -0,0 +1,33 @@
from __future__ import annotations
import jwt
from datetime import datetime, timedelta, timezone
from app.core.config import get_settings
def _now() -> datetime:
return datetime.now(timezone.utc)
def create_access_token(user_id: str, expires_minutes: int | None = None) -> str:
"""为指定用户签发 JWT 访问令牌。"""
settings = get_settings()
expire = _now() + timedelta(minutes=expires_minutes or settings.access_token_expire_minutes)
payload = {
"sub": user_id,
"iat": int(_now().timestamp()),
"exp": int(expire.timestamp()),
}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def decode_access_token(token: str) -> str | None:
"""校验并返回令牌中的用户 ID无效/过期返回 None。"""
settings = get_settings()
try:
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
except jwt.PyJWTError:
return None
sub = payload.get("sub")
return sub if isinstance(sub, str) else None