77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
# Test Router 页面
|
|
from typing import Optional
|
|
|
|
import aioredis
|
|
from fastapi import APIRouter, Depends, Query
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from fastapi_plugins import depends_redis
|
|
from fastapi_sqlalchemy import db
|
|
from sqlalchemy.sql import text
|
|
from fastapi_limiter.depends import RateLimiter
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from src.service.example import get_user_by_id
|
|
from src.dtos import response
|
|
from src.dtos.example import UserExampleListPagesResult
|
|
|
|
router = APIRouter()
|
|
|
|
# 并发数限制 每5秒最多100次请求
|
|
@router.get("/", dependencies=[Depends(RateLimiter(times=100, seconds=5))])
|
|
def index():
|
|
return {"msg": "This is Index Page"}
|
|
|
|
|
|
# ==================登录验证相关===================================
|
|
class User(BaseModel):
|
|
username: str
|
|
email: Optional[str] = None
|
|
full_name: Optional[str] = None
|
|
disabled: Optional[bool] = None
|
|
|
|
|
|
def fake_decode_token(token):
|
|
return User(
|
|
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
|
|
)
|
|
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
|
|
|
|
|
async def get_current_user(token: str = Depends(oauth2_scheme)):
|
|
print("get_current_user...")
|
|
user = fake_decode_token(token)
|
|
return user
|
|
|
|
@router.get('/user/{user_id}')
|
|
async def get_user(user_id: int):
|
|
return get_user_by_id(user_id=user_id)
|
|
|
|
# 登录验证
|
|
@router.get("/users/me")
|
|
async def read_users_me(current_user: User = Depends(get_current_user)):
|
|
return current_user
|
|
|
|
|
|
@router.get('/user_list_pages', response_model=UserExampleListPagesResult)
|
|
async def get_user_list_pages(page: int = Query(..., description="当前页码"),
|
|
limit: int = 20):
|
|
items = [{"id": 12, "user_name": "lala"}]
|
|
data = {"totalRecords": limit, "totalPages": 0, "currentPage": page, "items": items}
|
|
res = response(data=data, message="Ok!")
|
|
return res
|
|
|
|
# 数据库查询
|
|
@router.get('/get_db_version')
|
|
async def get_db_version():
|
|
result = db.session.execute(text("SELECT version()")).first()
|
|
return dict(result=result.tuple()[0])
|
|
|
|
# Redis 缓存查询
|
|
@router.get("/ping")
|
|
async def ping(cache: aioredis.Redis = Depends(depends_redis)):
|
|
return dict(ping=await cache.ping())
|
|
|