fastapi 简单使用

1. create database.py

from sqlalchemyimport create_engine

from sqlalchemy.ext.declarativeimport declarative_base

from sqlalchemy.ormimport sessionmaker

# SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

SQLALCHEMY_DATABASE_URL ="mysql://root:123456@localhost:3306/sqlalchemy_test"

# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(

SQLALCHEMY_DATABASE_URL

)

SessionLocal = sessionmaker(autocommit=False,autoflush=False,bind=engine)

Base = declarative_base()

2.  create crud.py

from sqlalchemyimport column

from sqlalchemy.ormimport Session

import models

import schemas

def get_user(db: Session, user_id:int):

return db.query(models.User).filter(column("id") == user_id).first()

def get_user_by_email(db: Session, email:str):

return db.query(models.User).filter(column("email") == email).first()

def get_users(db: Session, skip:int =0, limit:int =100):

return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):

fake_hashed_password = user.password +"notreallyhashed"

    db_user = models.User(email=user.email,hashed_password=fake_hashed_password)

db.add(db_user)

db.commit()

db.refresh(db_user)

return db_user

def get_items(db: Session, skip:int =0, limit:int =100):

return db.query(models.Item).offset(skip).limit(limit).all()

def create_user_item(db: Session, item: schemas.ItemCreate, user_id:int):

db_item = models.Item(**item.dict(),owner_id=user_id)

db.add(db_item)

db.commit()

db.refresh(db_item)

return db_item

3.  create models.py

from sqlalchemyimport Boolean, Column, ForeignKey, Integer, String,Text

from sqlalchemy.ormimport relationship

from databaseimport Base

class User(Base):

__tablename__ ="users"

    id = Column(Integer,primary_key=True,index=True)

email = Column(String(80),unique=True,index=True)

hashed_password = Column(String(100))

is_active = Column(Boolean,default=True)

items = relationship("Item",back_populates="owner")

class Item(Base):

__tablename__ ="items"

    id = Column(Integer,primary_key=True,index=True)

title = Column(String(255),index=True)

description = Column(Text)

owner_id = Column(Integer, ForeignKey("users.id"))

owner = relationship("User",back_populates="items")

4.  create schemas.py

from typingimport Union

from pydanticimport BaseModel

class ItemBase(BaseModel):

title:str

    description: Union[str,None] =None

class ItemCreate(ItemBase):

pass

class Item(ItemBase):

id:int

    owner_id:int

    class Config:

from_attributes =True

class UserBase(BaseModel):

email:str

class UserCreate(UserBase):

password:str

class User(UserBase):

id:int

    is_active:bool

    items:list[Item] = []

class Config:

from_attributes =True

5.  create main app

from fastapiimport Depends, FastAPI, HTTPException

from sqlalchemy.ormimport Session

import crud

import models

import schemas

from databaseimport SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency

def get_db():

db = SessionLocal()

try:

yield db

finally:

db.close()

@app.post("/users/",response_model=schemas.User)

def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):

db_user = crud.get_user_by_email(db,email=user.email)

if db_user:

raise HTTPException(status_code=400,detail="Email already registered")

create_user = crud.create_user(db=db,user=user)

print(type(create_user))

return create_user

@app.get("/users/",response_model=list[schemas.User])

def read_users(skip:int =0, limit:int =100, db: Session = Depends(get_db)):

users = crud.get_users(db,skip=skip,limit=limit)

return users

@app.get("/users/{user_id}",response_model=schemas.User)

def read_user(user_id:int, db: Session = Depends(get_db)):

db_user = crud.get_user(db,user_id=user_id)

if db_useris None:

raise HTTPException(status_code=404,detail="User not found")

return db_user

@app.post("/users/{user_id}/items/",response_model=schemas.Item)

def create_item_for_user(

user_id:int, item: schemas.ItemCreate, db: Session = Depends(get_db)

):

return crud.create_user_item(db=db,item=item,user_id=user_id)

@app.get("/items/",response_model=list[schemas.Item])

def read_items(skip:int =0, limit:int =100, db: Session = Depends(get_db)):

items = crud.get_items(db,skip=skip,limit=limit)

return items

注意点： @app.get("/users/",response_model=list[schemas.User])   

                @app.post("/users/",response_model=schemas.User) 

如果结果有多个对象，response_model 需要是 List

fastapi  数据库查询出来的 对象和 list<Obj>  可以正常序列化， 而 flask 和 django 都需要 转字典或者序列化器

比如：

@app.post("/students/", )# response_model=schemas.StudentDetail

def create_student(stu: schemas.StudentCreate, db: Session = Depends(get_db)):

db_stu = crud.get_student_by_name(db,user_name=stu.user_name)

if db_stu:

raise HTTPException(status_code=400,detail="user name already registered")

result = crud.create_stu(db=db,stu=stu)

return {'code':200,'data': result}

@app.get("/students/", )# response_model=list[schemas.StudentDetail]

def get_student(skip:int =0, limit:int =100, db: Session = Depends(get_db)):

sts = crud.get_students(db,skip=skip,limit=limit)

return {'code':200,'data': sts}