後端學習紀錄 Backend with FastAPI - Website with Departure

tags: backend

Copyright 2021, 月下麒麟

---

Target

該筆記主要目標為使用Web之Python框架FastAPI，
並結合資料庫框架Flask-SQLAlchemy與資料庫SQLite，
另外，會著重在前後端分離的應用，並以後端技術為主

Requirements

anyio==3.3.4
asgiref==3.4.1
click==8.0.3
colorama==0.4.4
fastapi==0.70.0
greenlet==1.1.2
h11==0.12.0
idna==3.3
Jinja2==3.0.2
MarkupSafe==2.0.1
pydantic==1.8.2
sniffio==1.2.0
SQLAlchemy==1.4.26
starlette==0.16.0
typing-extensions==3.10.0.2
uvicorn==0.15.0

Architecture

|---app
|    |---__init__.py
|    |---fastDB.db
|    |---database.py
|    |---model.py
|    |---schemas.py
|    |---crud.py 
|    |---main.py

Run Virtual Enviroment

關於創建與啟用虛擬，這邊就不再重複說明，可以參考我的另一篇後端學習紀錄 Backend with FastAPI - Setting Enviroment

Run ASGI Server

(app) D:\app> uvicorn main:app --reload
...
...
[32mINFO [0m:     Waiting for application startup.
[32mINFO [0m:     Application startup complete.
...
...

---

Source Code

Python
database主要在建立資料庫與物件對應的聯結，都繞乎於ORM

Reference:[Flask教學] Flask-SQLAlchemy 資料庫操作-ORM篇
Reference:SQLAlchemy 1.4 Documentation

# database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = r"sqlite:///./fastDB.db"

# create SQL engine
engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8', echo=True)

# create SQL communication session and bind
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# create SQL mapping table
Base = declarative_base()

---

剛才我們建立了與資料庫的連結，在model裡面則是建立資料庫表格的欄位

# model.py

# create model attribute/column
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Float

# relationship with ORM
#from sqlalchemy.orm import relationship

from sql_app.database import Base
from sqlalchemy.orm import relationship

class PeopleInfo(Base):
    __tablename__ = "humanInfo"
    id = Column(Integer, primary_key=True)
    name = Column(String(length=30))
    height = Column(Float)
    weight = Column(Float)
    habbit = Column(String(length=252))

對應真正的資料庫就會長這個樣子

---

在schemas則是會於產生物件時對其輸入資料屬性做檢查

# schemas.py

from typing import List
from pydantic import BaseModel

class PeopleBase(BaseModel):
    name: str
    height: float
    weight: float
    habbit: str

class PeopleUpdate(PeopleBase):
    id: int

    class Config:
        orm_mode = True

class PeopleType(BaseModel):
    skip: int
    limit: int
    data: List[PeopleUpdate]

---

這邊就是對資料庫的操作行為
get(Create), post(Read), put(Update, delete(Delete)

# crud.py

from typing import List
from fastapi.exceptions import HTTPException
from sql_app import model
from sqlalchemy.orm import Session, session
from sql_app.model import PeopleInfo
from sql_app.schemas import PeopleBase, PeopleUpdate, PeopleType

def get_user_by_id(boxSession: Session, _id: int):
    return boxSession.query(PeopleInfo).filter(PeopleInfo.id == _id).first()

def get_users(boxSession: Session, _skip: int=0, _limit: int=100) -> List[PeopleInfo]:
    return boxSession.query(PeopleInfo).offset(_skip).limit(_limit).all()

def create_user(boxSession: Session, _createData: PeopleUpdate) -> PeopleInfo :
    peopleDetail = boxSession.query(PeopleInfo).filter(PeopleInfo.name == _createData.name,
                                                        PeopleInfo.height == _createData.height,
                                                        PeopleInfo.weight == _createData.weight,
                                                        PeopleInfo.habbit == _createData.habbit).first()
    if peopleDetail is not None:
        raise HTTPException(status_code=409, detail="People already exist.")
    
    newPeopleInfo = PeopleInfo(**_createData.dict())
    boxSession.add(newPeopleInfo)
    boxSession.commit()
    boxSession.refresh(newPeopleInfo)
    return newPeopleInfo

def update_user(boxSession: Session, _id: int , infoUpdate: PeopleUpdate) -> PeopleInfo:
    orignalInfo = get_user_by_id(boxSession, _id)

    if orignalInfo is None:
        raise HTTPException(status_code=404 , detail="404 Not Found.")
    
    orignalInfo.name = infoUpdate.name
    orignalInfo.height = infoUpdate.height
    orignalInfo.weight = infoUpdate.weight
    orignalInfo.habbit = infoUpdate.habbit
    boxSession.commit()
    boxSession.refresh(orignalInfo)

    return orignalInfo


def delete_user(boxSession: Session, _id: int):
    res = get_user_by_id(boxSession, _id)
    if res is None:
        raise HTTPException(status_code=404, detail="404 Not Found.")
    boxSession.delete(res)
    boxSession.commit()
    return { "code": 0 }

---

在ASGI Server啟動後會執行該main函式，所以這邊也定義了對網頁行為的操作
Get, Post, Put, Delete

# main.py

from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware

from sql_app import crud, model
from sql_app.model import PeopleInfo
from sql_app.schemas import PeopleBase, PeopleType, PeopleUpdate
from sql_app.database import SessionLocal, engine
from sqlalchemy.orm.session import Session

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

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# -----------------------------------
origins = ["http://localhost:8000",
            "http://localhost:3000"]

app.add_middleware(CORSMiddleware,
                    allow_origins = origins,
                    allow_credentials = True,
                    allow_methods=["*"],
                    allow_headers = ["*"])
# -----------------------------------
@app.get("/users/{user_id}",response_model=PeopleUpdate)
def read_user_id(user_id: int, db:Session= Depends(get_db)):
    res = crud.get_user_by_id(db,user_id)
    if res is None:
        raise HTTPException(status_code=404, detail="404 Not Found.")
    return res

@app.get("/users",response_model=PeopleType)
def read_users(skip: int=0, limit: int=100, db:Session= Depends(get_db)):
    res = crud.get_users(db, skip, limit)
    response = {"skip":skip , "limit":limit , "data":res}
    return response

@app.post("/users",response_model=PeopleUpdate)
def create_user(userForm: PeopleUpdate, db:Session= Depends(get_db)):
    try:
        res = crud.create_user(db, userForm)
        return res
    except Exception as err:
        return HTTPException(**err.__dict__)

@app.put("/users/{user_id}", response_model=PeopleUpdate)
def update_user(userForm: PeopleUpdate, user_id: int, db:Session= Depends(get_db)):
    try:
        res = crud.update_user(db, user_id, userForm)
        return res
    except Exception as err:
        raise HTTPException(**err.__dict__)

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db:Session= Depends(get_db)):
    try:
        crud.delete_user(db,user_id)
    except Exception as err:
        raise HTTPException(status_code=404, detail="404 Not Found.")
    return {"code": 0}

Testing

Open FastAPI Document

run server後

開啟網頁

做CRUD的測試，這個API測試文檔是fastAPI自帶的
只要在port後面加入/docs路由就可開啟使用

舉例:
對這支API測試讀取資料庫裏面的第6筆資料

如果有取得成功，Server就會回覆內容與status code

對應資料庫data是一樣的

其他的API testing也是相似的方法，就不再舉例。

---

Summary

這算是自我練習比較接近前後端分離，後端的架構應用
所以，當你再去閱讀我的前端筆記，就可以將前後端接起來了(工商一下)

• Day1學前端 Frontend with React - Start React
• Day2學前端 Frontend with React - Hello World React
• Day3學前端 Frontend with React - 串接API with Ajax
• Day4學前端 Frontend with React - show on UI

最後，關乎於將database, model, crud, schemas, main的Python檔分門別類
是參考以下的兩個連結去改寫的，當中有一些類別的應用與ORM的對應
之後，有機會再來寫一篇關乎於物件的探討。
reference: SQL (Relational) Databases
reference: Building a CRUD APP with FastAPI and MySQL

謝謝您的收看，歡迎留言一起探討~