fastapi-extend 0.1.1

fastapi extend

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)