fastapi extend
Project description
1、install
pip install fastapi_extend
2、use
from sqlalchemy import Column, String, Integer, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base # ORM(对象关系映射)的基类
Base = declarative_base()
class Trade(Base):
__tablename__ = "trade"
id = Column(Integer, primary_key=True)
account = Column(String(20), default=None, nullable=False)
trade_datetime = Column(DateTime)
volume = Column(Float, default=None, nullable=False)
price = Column(Float, default=0, nullable=False)
order_id = Column(String(20))
1)、 jwt
from pydantic import BaseModel
from fastapi_extend import AuthHandler
from fastapi import APIRouter, Depends
auth_handler = AuthHandler()
class AuthDetails(BaseModel):
username: str
password: str
router = APIRouter()
@router.post('/login')
async def login(auth_details: AuthDetails):
token = auth_handler.encode_token(auth_details.username)
return {'data': token, "status": "success"}
@router.get('/protected')
async def protected(username=Depends(auth_handler.auth_wrapper)):
return {'name': username}
2)、Pagination
from fastapi_extend.pagenator import PageNumberPagination
from fastapi_extend.serializer import Serializer
from pydantic import validator
from sqlalchemy import Column, String, Integer, Float, DateTime
class TradeSer(Serializer):
account: str
trade_datetime = Column(DateTime)
volume: float
price: float
order_id: str
paginator = PageNumberPagination(
params,
Trade,
TradeSer,
exclude={"start", "end"},
)
query = paginator.get_queryset()
if params.start:
query = query.where(Trade.trade_datetime >= params.start)
if params.end:
query = query.where(Trade.trade_datetime <= params.end)
data = await paginator.paginate_query(query, session)
3)、Serializer
from fastapi_extend.serializer import Serializer
from pydantic import validator
class TradeSer(Serializer):
account: str
volume: float
price: float
order_id: str
@validator("price")
def price_add(cls, v):
return v + 10
data_ser = TradeSer.dump(result.scalars(), many=True)