แอพประสิทธิภาพสูงพร้อม Python – บทช่วยสอน FastAPI

กรอบงานภาษาโปรแกรมที่ดีทำให้ง่ายต่อการผลิตผลิตภัณฑ์ที่มีคุณภาพเร็วขึ้น กรอบงานที่ยอดเยี่ยมทำให้ประสบการณ์การพัฒนาทั้งหมดเป็นเรื่องสนุก FastAPI คือเฟรมเวิร์กเว็บ Python ใหม่ที่ทรงพลังและน่าใช้ คุณสมบัติต่อไปนี้ทำให้ FastAPI คุ้มค่าที่จะลอง:

• ความเร็ว: FastAPI เป็นหนึ่งในเฟรมเวิร์กเว็บ Python ที่เร็วที่สุด อันที่จริง ความเร็วของมันเทียบเท่ากับ Node.js และ Go ตรวจสอบการทดสอบประสิทธิภาพเหล่านี้
• เอกสารนักพัฒนาโดยละเอียดและใช้งานง่าย
• พิมพ์คำใบ้รหัสของคุณและรับการตรวจสอบและแปลงข้อมูลฟรี
• สร้างปลั๊กอินอย่างง่ายดายโดยใช้การฉีดพึ่งพา

การสร้างแอป TODO

หากต้องการสำรวจแนวคิดใหญ่เบื้องหลัง FastAPI ให้สร้างแอป TODO ซึ่งตั้งค่ารายการสิ่งที่ต้องทำสำหรับผู้ใช้ แอพเล็ก ๆ ของเราจะมีคุณสมบัติดังต่อไปนี้:

• สมัครสมาชิกและเข้าสู่ระบบ
• เพิ่มรายการสิ่งที่ต้องทำใหม่
• รับรายการสิ่งที่ต้องทำทั้งหมด
• ลบ/อัปเดตรายการสิ่งที่ต้องทำ

SQLAlchemy สำหรับโมเดลข้อมูล

แอพของเรามีเพียงสองรุ่น: ผู้ใช้และสิ่งที่ต้องทำ ด้วยความช่วยเหลือของ SQLAlchemy ซึ่งเป็นชุดเครื่องมือฐานข้อมูลสำหรับ Python เราสามารถแสดงแบบจำลองของเราดังนี้:

 class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) lname = Column(String) fname = Column(String) email = Column(String, unique=True, index=True) todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan") class TODO(Base): __tablename__ = "todos" id = Column(Integer, primary_key=True, index=True) text = Column(String, index=True) completed = Column(Boolean, default=False) owner_id = Column(Integer, ForeignKey("users.id")) owner = relationship("User", back_populates="todos")

เมื่อโมเดลของเราพร้อมแล้ว มาเขียนไฟล์กำหนดค่าสำหรับ SQLAlchemy เพื่อให้รู้วิธีสร้างการเชื่อมต่อกับฐานข้อมูล

 import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL'] engine = create_engine( SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

ปลดปล่อยพลังของคำแนะนำประเภท

ส่วนที่มีขนาดใหญ่ของโครงการ API ใดๆ เกี่ยวข้องกับงานประจำ เช่น การตรวจสอบความถูกต้องของข้อมูลและการแปลง มาจัดการกันก่อนที่เราจะเขียนตัวจัดการคำขอ ด้วย FastAPI เราแสดงสคีมาของข้อมูลขาเข้า/ขาออกของเราโดยใช้โมเดล pydantic จากนั้นใช้โมเดล pydantic เหล่านี้เพื่อพิมพ์คำใบ้และเพลิดเพลินไปกับการตรวจสอบและการแปลงข้อมูลฟรี โปรดทราบว่าโมเดลเหล่านี้ไม่เกี่ยวข้องกับเวิร์กโฟลว์ฐานข้อมูลของเรา และระบุเฉพาะรูปร่างของข้อมูลที่ไหลเข้าและออกจากอินเทอร์เฟซ REST ของเรา ในการเขียนแบบจำลอง pydantic ให้นึกถึงวิธีที่ข้อมูลผู้ใช้และสิ่งที่ต้องทำจะไหลเข้าและออก

ตามเนื้อผ้า ผู้ใช้ใหม่จะสมัครใช้บริการ TODO ของเรา และผู้ใช้ที่มีอยู่จะเข้าสู่ระบบ การโต้ตอบทั้งสองนี้จะจัดการกับข้อมูลผู้ใช้ แต่รูปร่างของข้อมูลจะแตกต่างกัน เราต้องการข้อมูลเพิ่มเติมจากผู้ใช้ในระหว่างการสมัครและขั้นต่ำ (เฉพาะอีเมลและรหัสผ่าน) เมื่อเข้าสู่ระบบ ซึ่งหมายความว่าเราต้องการโมเดล pydantic สองแบบเพื่อแสดงรูปร่างที่แตกต่างกันของข้อมูลผู้ใช้ทั้งสองนี้

อย่างไรก็ตาม ในแอป TODO ของเรา เราจะใช้ประโยชน์จากการสนับสนุน OAuth2 ในตัวใน FastAPI สำหรับขั้นตอนการเข้าสู่ระบบที่ใช้ JSON Web Token (JWT) เราเพียงแค่ต้องกำหนด UserCreate schema ที่นี่เพื่อระบุข้อมูลที่จะไหลเข้าสู่ปลายทางการลงชื่อสมัครใช้ของเรา และ UserBase schema เพื่อส่งคืนเป็นการตอบสนองในกรณีที่กระบวนการสมัครสำเร็จ

 from pydantic import BaseModel from pydantic import EmailStr class UserBase(BaseModel): email: EmailStr class UserCreate(UserBase): lname: str fname: str password: str

ในที่นี้ เราทำเครื่องหมายนามสกุล ชื่อจริง และรหัสผ่านเป็นสตริง แต่สามารถทำให้รัดกุมยิ่งขึ้นได้โดยใช้สตริงที่มีข้อจำกัด pydantic ที่เปิดใช้งานการตรวจสอบ เช่น ความยาวต่ำสุด ความยาวสูงสุด และ regexes

เพื่อสนับสนุนการสร้างและแสดงรายการ TODO เรากำหนดสคีมาต่อไปนี้:

 class TODOCreate(BaseModel): text: str completed: bool

เพื่อสนับสนุนการอัปเดตรายการสิ่งที่ต้องทำที่มีอยู่ เราได้กำหนดสคีมาอื่น:

 class TODOUpdate(TODOCreate): id: int

ด้วยเหตุนี้ เราจึงเสร็จสิ้นการกำหนดสคีมาสำหรับการแลกเปลี่ยนข้อมูลทั้งหมด ตอนนี้เราหันความสนใจไปที่ตัวจัดการคำขอซึ่งสคีมาเหล่านี้จะถูกนำมาใช้เพื่อดำเนินการแปลงข้อมูลและตรวจสอบความถูกต้องทั้งหมดโดยไม่เสียค่าใช้จ่าย

ให้ผู้ใช้ลงทะเบียน

ขั้นแรก ให้อนุญาตให้ผู้ใช้ลงชื่อสมัครใช้ เนื่องจากบริการทั้งหมดของเราจำเป็นต้องได้รับการเข้าถึงโดยผู้ใช้ที่ได้รับการรับรองความถูกต้อง เราเขียนตัวจัดการคำขอแรกของเราโดยใช้สคีมา UserCreate และ UserBase ที่กำหนดไว้ด้านบน

 @app.post("/api/users", response_model=schemas.User) def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)): """add new user""" user = crud.get_user_by_email(db, user_data.email) if user: raise HTTPException(status_code=409, detail="Email already registered.") signedup_user = crud.create_user(db, user_data) return signedup_user

มีหลายอย่างเกิดขึ้นในโค้ดสั้นๆ นี้ เราได้ใช้มัณฑนากรเพื่อระบุกริยา HTTP, URI และสคีมาของการตอบสนองที่ประสบความสำเร็จ เพื่อให้แน่ใจว่าผู้ใช้ส่งข้อมูลที่ถูกต้อง เราได้พิมพ์คำใบ้เนื้อหาคำขอด้วยสคีมา UserCreate ที่กำหนดไว้ก่อนหน้านี้ เมธอดนี้กำหนดพารามิเตอร์อื่นสำหรับการจัดการฐานข้อมูล—นี่คือการแทรกการพึ่งพาในการดำเนินการ และจะกล่าวถึงในภายหลังในบทช่วยสอนนี้

การรักษาความปลอดภัย API ของเรา

เราต้องการคุณสมบัติความปลอดภัยต่อไปนี้ในแอพของเรา:

• การแฮชรหัสผ่าน
• การรับรองความถูกต้องตาม JWT

สำหรับการแฮชรหัสผ่าน เราสามารถใช้ Passlib มากำหนดฟังก์ชันที่จัดการการแฮชรหัสผ่านและตรวจสอบว่ารหัสผ่านถูกต้องหรือไม่

 from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def authenticate_user(db, email: str, password: str): user = crud.get_user_by_email(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user

ในการเปิดใช้งานการพิสูจน์ตัวตนแบบ JWT เราจำเป็นต้องสร้าง JWT และถอดรหัสเพื่อรับข้อมูลรับรองผู้ใช้ เรากำหนดฟังก์ชันต่อไปนี้เพื่อให้ฟังก์ชันนี้

 # install PyJWT import jwt from fastapi.security import OAuth2PasswordBearer SECRET_KEY = os.environ['SECRET_KEY'] ALGORITHM = os.environ['ALGORITHM'] def create_access_token(*, data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def decode_access_token(db, token): credentials_exception = HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get("sub") if email is None: raise credentials_exception token_data = schemas.TokenData(email=email) except PyJWTError: raise credentials_exception user = crud.get_user_by_email(db, email=token_data.email) if user is None: raise credentials_exception return user

ออกโทเค็นเมื่อเข้าสู่ระบบสำเร็จ

ตอนนี้ เราจะกำหนดปลายทางการเข้าสู่ระบบและใช้ขั้นตอนรหัสผ่าน OAuth2 ปลายทางนี้จะได้รับอีเมลและรหัสผ่าน เราจะตรวจสอบข้อมูลประจำตัวกับฐานข้อมูล และเมื่อสำเร็จ จะออกโทเค็นเว็บ JSON ให้กับผู้ใช้

ในการรับข้อมูลประจำตัว เราจะใช้ OAuth2PasswordRequestForm ซึ่งเป็นส่วนหนึ่งของยูทิลิตี้ความปลอดภัยของ FastAPI

 @app.post("/api/token", response_model=schemas.Token) def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()): """generate access token for valid credentials""" user = authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token(data={"sub": user.email}, expires_delta=access_token_expires) return {"access_token": access_token, "token_type": "bearer"}

การใช้การพึ่งพาการฉีดเพื่อเข้าถึงฐานข้อมูลและป้องกันปลายทาง

เราได้ตั้งค่าปลายทางการเข้าสู่ระบบที่จัดเตรียม JWT ให้กับผู้ใช้เมื่อเข้าสู่ระบบสำเร็จ ผู้ใช้สามารถบันทึกโทเค็นนี้ในที่จัดเก็บในตัวเครื่องและแสดงให้ส่วนหลังของเราเป็นส่วนหัวการให้สิทธิ์ ปลายทางที่คาดหวังการเข้าถึงจากผู้ใช้ที่เข้าสู่ระบบเท่านั้นสามารถถอดรหัสโทเค็นและค้นหาว่าใครเป็นผู้ร้องขอ งานประเภทนี้ไม่ได้ผูกติดอยู่กับปลายทางเฉพาะ แต่เป็นตรรกะที่ใช้ร่วมกันที่ใช้ในปลายทางที่ได้รับการป้องกันทั้งหมด ทางที่ดีควรตั้งค่าตรรกะการถอดรหัสโทเค็นเป็นการพึ่งพาที่สามารถใช้ในตัวจัดการคำขอใดก็ได้

ในการพูด FastAPI ฟังก์ชันการดำเนินการพาธของเรา (ตัวจัดการคำขอ) จะขึ้นอยู่กับ get_current_user การพึ่งพา get_current_user จำเป็นต้องมีการเชื่อมต่อกับฐานข้อมูลและเชื่อมต่อกับตรรกะ OAuth2PasswordBearer ของ OAuth2PasswordBearer เพื่อรับโทเค็น เราจะแก้ไขปัญหานี้โดยทำให้ get_current_user ขึ้นอยู่กับฟังก์ชันอื่นๆ ด้วยวิธีนี้ เราสามารถกำหนดห่วงโซ่การพึ่งพาซึ่งเป็นแนวคิดที่ทรงพลังมาก

 def get_db(): """provide db session to path operation functions""" try: db = SessionLocal() yield db finally: db.close() def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): return decode_access_token(db, token) @app.get("/api/me", response_model=schemas.User) def read_logged_in_user(current_user: models.User = Depends(get_current_user)): """return user settings for current user""" return current_user

ผู้ใช้ที่เข้าสู่ระบบสามารถ CRUD TODOs

ก่อนที่เราจะเขียนฟังก์ชันการทำงานของพาธสำหรับ TODO Create, Read, Update, Delete (CRUD) เราจะกำหนดฟังก์ชันตัวช่วยต่อไปนี้เพื่อดำเนินการ CRUD จริงบน db

 def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate): todo = models.TODO(text=todo_data.text, completed=todo_data.completed) todo.owner = current_user db.add(todo) db.commit() db.refresh(todo) return todo def update_todo(db: Session, todo_data: schemas.TODOUpdate): todo = db.query(models.TODO).filter(models.TODO.id == id).first() todo.text = todo_data.text todo.completed = todo.completed db.commit() db.refresh(todo) return todo def delete_todo(db: Session, id: int): todo = db.query(models.TODO).filter(models.TODO.id == id).first() db.delete(todo) db.commit() def get_user_todos(db: Session, userid: int): return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

ฟังก์ชันระดับ db เหล่านี้จะใช้ในปลายทาง REST ต่อไปนี้:

 @app.get("/api/mytodos", response_model=List[schemas.TODO]) def get_own_todos(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """return a list of TODOs owned by current user""" todos = crud.get_user_todos(db, current_user.id) return todos @app.post("/api/todos", response_model=schemas.TODO) def add_a_todo(todo_data: schemas.TODOCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """add a TODO""" todo = crud.create_meal(db, current_user, meal_data) return todo @app.put("/api/todos/{todo_id}", response_model=schemas.TODO) def update_a_todo(todo_id: int, todo_data: schemas.TODOUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """update and return TODO for given id""" todo = crud.get_todo(db, todo_id) updated_todo = crud.update_todo(db, todo_id, todo_data) return updated_todo @app.delete("/api/todos/{todo_id}") def delete_a_meal(todo_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """delete TODO of given id""" crud.delete_meal(db, todo_id) return {"detail": "TODO Deleted"}

เขียนแบบทดสอบ

มาเขียนการทดสอบสองสามอย่างสำหรับ TODO API ของเรากัน FastAPI มีคลาส TestClient ที่อิงตามไลบรารี Requests ยอดนิยม และเราสามารถรันการทดสอบด้วย Pytest

เพื่อให้แน่ใจว่ามีเพียงผู้ใช้ที่เข้าสู่ระบบเท่านั้นที่สามารถสร้าง TODO ได้ เราสามารถเขียนดังนี้:

 from starlette.testclient import TestClient from .main import app client = TestClient(app) def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False) response = client.post("/api/todos", data=todo) assert response.status_code == 401

การทดสอบต่อไปนี้จะตรวจสอบปลายทางการเข้าสู่ระบบของเราและสร้าง JWT หากแสดงข้อมูลรับรองการเข้าสู่ระบบที่ถูกต้อง

 def test_user_can_obtain_auth_token(): response = client.post("/api/token", data=good_credentials) assert response.status_code == 200 assert 'access_token' in response.json() assert 'token_type' in response.json()

บทสรุป

เสร็จสิ้นการติดตั้งแอป TODO แบบง่ายๆ โดยใช้ FastAPI ถึงตอนนี้ คุณได้เห็นพลังของคำแนะนำประเภทที่นำไปใช้ได้ดีในการกำหนดรูปร่างของข้อมูลขาเข้าและขาออกผ่านอินเทอร์เฟซ REST ของเรา เรากำหนดสคีมาไว้ที่เดียวและปล่อยให้ FastAPI ใช้การตรวจสอบและการแปลงข้อมูล คุณลักษณะเด่นอื่น ๆ คือการแทรกการพึ่งพา เราใช้แนวคิดนี้เพื่อจัดแพ็คเกจตรรกะที่ใช้ร่วมกันในการรับการเชื่อมต่อฐานข้อมูล ถอดรหัส JWT เพื่อรับผู้ใช้ที่เข้าสู่ระบบอยู่ในปัจจุบัน และใช้ OAuth2 อย่างง่ายด้วยรหัสผ่านและผู้ถือ นอกจากนี้เรายังเห็นว่าการพึ่งพาสามารถเชื่อมโยงเข้าด้วยกันได้อย่างไร

เราสามารถนำแนวคิดนี้ไปใช้เพื่อเพิ่มคุณสมบัติ เช่น การเข้าถึงตามบทบาทได้อย่างง่ายดาย นอกจากนี้ เรากำลังเขียนโค้ดที่กระชับและมีประสิทธิภาพโดยไม่ต้องเรียนรู้ลักษณะเฉพาะของเฟรมเวิร์ก พูดง่ายๆ ก็คือ FastAPI คือชุดเครื่องมืออันทรงพลังที่คุณไม่ต้องเรียนรู้เพราะเป็นเครื่องมือ Python ที่ทันสมัย มีความสุข.