Fastapi

日志重写 import logging import os import sys from pprint import pformat from loguru import logger from config import settings class InterceptHandler(logging.Handler): def emit(self, record): # Get corresponding Loguru level if it exists try: level = logger.level(record.levelname).name except ValueError: level = record.levelno # Find caller from where originated the logged message frame, depth = logging.currentframe(), 2 while frame.f_code.co_filename == logging.__file__: frame = frame.f_back depth += 1 logger.opt(depth=depth, exception=record.exc_info).log( level, record.getMessage() ) def format_record(record: dict) -> str: format_string = settings.config.LOG_FORMAT if record["extra"].get("payload") is not None: record["extra"]["payload"] = pformat( record["extra"]["payload"], indent=4, compact=True, width=88 ) format_string += "\n<level>{extra[payload]}</level>" format_string += "{exception}\n" return format_string # it requires app init or won't be work def logger_init(): log_path_all = os.path.join(settings.config.log_path, settings.config.LOG_FILE) log_path_error = os.path.join(settings.config.log_error_path, settings.config.LOG_FILE) logging.getLogger().handlers = [InterceptHandler()] logger.configure( handlers=[{"sink": sys.stdout, "level": logging.DEBUG, "format": format_record}]) # logger.add(config.LOG_DIR, encoding='utf-8', rotation="9:46") logger.add(log_path_all, rotation="00:00", retention="2 week", enqueue=True, diagnose=True, level="DEBUG", format=format_record ) logger.add(log_path_error, rotation="00:00", retention="3 week", enqueue=True, backtrace=True, diagnose=True, level="ERROR", format=format_record ) logger.debug('日志系统已加载\n\n\n') logging.getLogger("uvicorn.access").handlers = [InterceptHandler()] 在app中调用 app = FastAPI( title=config.settings.title, debug=config.settings.config.DEBUG ) # app create frist, then init logger logger_init() origins = [ "http://127.0.0.1", f"http://127.0.0.1:{config.settings.config.port}", "http://localhost", f"http://localhost:{config.settings.config.port}", # "*" ] app.include_router(scraper.router) app.include_router(works.router) app.include_router(users.router) app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) if __name__ == '__main__': uvicorn.run( app='main:app', host='0.0.0.0', port=config.settings.config.port, reload=True, # should be False at production access_log=True, ws_ping_interval=300, ws_ping_timeout=300 )

以Depends的方式定义jwt from jose import JWTError, jwt from passlib.context import CryptContext from fastapi.security import OAuth2PasswordBearer from fastapi import Depends, HTTPException, status from db import Curd from typing import Optional from datetime import datetime from config import settings from utils import logger from .db import get_db from datetime import timedelta pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/login") def hash_password(plain_password: str) -> str: return pwd_context.hash(plain_password) def verity_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) async def jwt_authenticate_user(username: str, password: str): db = get_db() user = await Curd.get_user_by_name(await db.__anext__(), username) if not user: return False if not verity_password(password, user.password): return False return user def created_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.config.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(claims=to_encode, key=settings.default.SECRET_KEY, algorithm="HS256") return encoded_jwt async def jwt_verity(token: str = Depends(oauth2_schema)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: playload = jwt.decode(token=token, key=settings.default.SECRET_KEY, algorithms=["HS256"]) except JWTError as e: logger.error("验证出现错误:\n", e) raise credentials_exception else: return playload async def jwt_get_current_user(playload: dict = Depends(jwt_verity)): db = get_db() user_id = int(playload.get("sub")) user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id) return user async def jwt_verity_admin(playload: dict = Depends(jwt_verity)): db = get_db() user_id = int(playload.get("sub")) is_admin = playload.get("is_admin") if is_admin: user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id) if user.is_admin: return raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) 需要分发token的url router = APIRouter( prefix="/users", tags=["user"], ) # it defind at dependencies.jwt # oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/token") @router.post("/register", response_model=UserToken) async def user_register(user: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)): if await Curd.is_user_name_exists(db, user.username): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="username already exists") hashed_password = hash_password(user.password) new_user = UserCreate(id=worker.get_id(), name=user.username, password=hashed_password) await Curd.register_user(db, new_user) @router.post("/login", response_model=UserToken) async def user_login(user: OAuth2PasswordRequestForm = Depends()): user = await jwt_authenticate_user(username=user.username, password=user.password) if not user: raise HTTPException( status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token = created_access_token(data={"sub": str(user.id)}) user_token = UserToken(username=user.name, access_token=access_token, token_type="bearer") return user_token 需要token进行访问限制的url router = APIRouter( prefix="/works", tags=["works"], ) @router.get("/get", response_model=Works) async def test( order: int, nsfw: bool, page: int, sort: int, user: User = Depends(jwt_get_current_user), db: AsyncSession = Depends(get_db), ): works = await Curd.get_works(db, user.id, order, nsfw, page, sort) return works

轻量web服务 介绍 FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 3.6+ 并基于标准的 Python 类型提示。 FastAPI 框架，高性能，易于学习，高效编码，生产可用 ...