Aplikasi Berperforma Tinggi dengan Python – Tutorial FastAPI

Diterbitkan: 2022-03-11

Kerangka kerja bahasa pemrograman yang baik memudahkan untuk menghasilkan produk berkualitas lebih cepat. Kerangka kerja yang hebat bahkan membuat seluruh pengalaman pengembangan menjadi menyenangkan. FastAPI adalah kerangka kerja web Python baru yang kuat dan menyenangkan untuk digunakan. Fitur-fitur berikut membuat FastAPI layak untuk dicoba:

  • Kecepatan: FastAPI adalah salah satu kerangka kerja web Python tercepat. Faktanya, kecepatannya setara dengan Node.js dan Go. Periksa tes kinerja ini.
  • Dokumen pengembang yang terperinci dan mudah digunakan
  • Ketik petunjuk kode Anda dan dapatkan validasi dan konversi data gratis.
  • Buat plugin dengan mudah menggunakan injeksi ketergantungan.

Membangun Aplikasi TODO

Untuk menjelajahi ide besar di balik FastAPI, mari buat aplikasi TODO, yang menyiapkan daftar tugas untuk penggunanya. Aplikasi kecil kami akan menyediakan fitur-fitur berikut:

  • Daftar dan Masuk
  • Tambahkan item TODO baru
  • Dapatkan daftar semua TODO
  • Hapus/Perbarui item TODO

SQLAlchemy untuk Model Data

Aplikasi kami hanya memiliki dua model: Pengguna dan TODO. Dengan bantuan SQLAlchemy, toolkit database untuk Python, kita dapat mengekspresikan model kita seperti ini:

 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")

Setelah model kita siap, mari kita tulis file konfigurasi untuk SQLAlchemy agar tahu cara membuat koneksi dengan database.

 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()

Melepaskan Kekuatan Petunjuk Tipe

Bagian yang cukup besar dari setiap proyek API menyangkut hal-hal rutin seperti validasi data dan konversi. Mari kita atasi terlebih dahulu sebelum kita menulis penangan permintaan. Dengan FastAPI, kami mengekspresikan skema data masuk/keluar kami menggunakan model pydantic dan kemudian menggunakan model pydantic ini untuk mengetik petunjuk dan menikmati validasi dan konversi data gratis. Harap dicatat model ini tidak terkait dengan alur kerja database kami dan hanya menentukan bentuk data yang mengalir masuk dan keluar dari antarmuka REST kami. Untuk menulis model pydantic, pikirkan semua cara informasi Pengguna dan TODO akan mengalir masuk dan keluar.

Biasanya, pengguna baru akan mendaftar ke layanan TODO kami dan pengguna yang sudah ada akan masuk. Kedua interaksi ini berhubungan dengan informasi Pengguna, tetapi bentuk datanya akan berbeda. Kami membutuhkan lebih banyak informasi dari pengguna saat mendaftar dan minimal (hanya email dan kata sandi) saat masuk. Ini berarti kami membutuhkan dua model pydantic untuk mengekspresikan dua bentuk berbeda dari info Pengguna ini.

Namun, di aplikasi TODO kami, kami akan memanfaatkan dukungan OAuth2 bawaan di FastAPI untuk alur masuk berbasis JSON Web Tokens (JWT). Kami hanya perlu mendefinisikan skema UserCreate di sini untuk menentukan data yang akan mengalir ke titik akhir pendaftaran kami dan skema UserBase untuk kembali sebagai respons jika proses pendaftaran berhasil.

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

Di sini, kami menandai nama belakang, nama depan, dan kata sandi sebagai string, tetapi dapat lebih diperketat dengan menggunakan string dibatasi pydantic yang memungkinkan pemeriksaan seperti panjang min, panjang maksimal, dan regex.

Untuk mendukung pembuatan dan daftar item TODO, kami mendefinisikan skema berikut:

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

Untuk mendukung pembaruan item TODO yang ada, kami mendefinisikan skema lain:

 class TODOUpdate(TODOCreate): id: int

Dengan ini, kita selesai dengan mendefinisikan skema untuk semua pertukaran data. Kami sekarang mengalihkan perhatian kami ke penangan permintaan di mana skema ini akan digunakan untuk melakukan semua pekerjaan berat konversi dan validasi data secara gratis.

Biarkan Pengguna Mendaftar

Pertama, izinkan pengguna untuk mendaftar, karena semua layanan kami harus diakses oleh pengguna yang diautentikasi. Kami menulis penangan permintaan pertama kami menggunakan skema UserCreate dan UserBase ditentukan di atas.

 @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

Ada banyak hal yang terjadi dalam kode pendek ini. Kami telah menggunakan dekorator untuk menentukan kata kerja HTTP, URI, dan skema respons yang berhasil. Untuk memastikan bahwa pengguna telah mengirimkan data yang benar, kami telah mengetikkan petunjuk pada badan permintaan dengan skema UserCreate yang ditentukan sebelumnya. Metode ini mendefinisikan parameter lain untuk mendapatkan pegangan pada database—ini adalah injeksi ketergantungan yang sedang beraksi dan akan dibahas nanti dalam tutorial ini.

Mengamankan API Kami

Kami menginginkan fitur keamanan berikut di aplikasi kami:

  • Hashing kata sandi
  • Otentikasi berbasis JWT

Untuk hashing kata sandi, kita dapat menggunakan Passlib. Mari kita definisikan fungsi yang menangani hashing kata sandi dan memeriksa apakah kata sandi sudah benar.

 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

Untuk mengaktifkan otentikasi berbasis JWT, kita perlu membuat JWT serta mendekodekannya untuk mendapatkan kredensial pengguna. Kami mendefinisikan fungsi berikut untuk menyediakan fungsi ini.

 # 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

Keluarkan Token saat Login Berhasil

Sekarang, kita akan mendefinisikan titik akhir Login dan mengimplementasikan alur kata sandi OAuth2. Endpoint ini akan menerima email dan password. Kami akan memeriksa kredensial terhadap database, dan jika berhasil, mengeluarkan token web JSON kepada pengguna.

Untuk menerima kredensial, kami akan menggunakan OAuth2PasswordRequestForm , yang merupakan bagian dari utilitas keamanan 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"}

Menggunakan Injeksi Ketergantungan untuk Mengakses DB dan Melindungi Titik Akhir

Kami telah menyiapkan titik akhir login yang menyediakan JWT kepada pengguna setelah login berhasil. Pengguna dapat menyimpan token ini di penyimpanan lokal dan menunjukkannya ke bagian belakang kami sebagai header Otorisasi. Titik akhir yang mengharapkan akses hanya dari pengguna yang masuk dapat memecahkan kode token dan mencari tahu siapa pemohonnya. Pekerjaan semacam ini tidak terikat pada titik akhir tertentu, melainkan logika bersama yang digunakan di semua titik akhir yang dilindungi. Yang terbaik adalah menyiapkan logika token-decoding sebagai dependensi yang dapat digunakan di penangan permintaan apa pun.

Dalam bahasa FastAPI, fungsi operasi jalur kami (penangan permintaan) kemudian akan bergantung pada get_current_user . Ketergantungan get_current_user harus memiliki koneksi ke database dan terhubung ke logika OAuth2PasswordBearer untuk mendapatkan token. Kami akan mengatasi masalah ini dengan membuat get_current_user bergantung pada fungsi lain. Dengan cara ini, kita dapat mendefinisikan rantai ketergantungan, yang merupakan konsep yang sangat kuat.

 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

Pengguna yang Masuk Dapat CRUD TODOs

Sebelum kita menulis fungsi operasi path untuk TODO Create, Read, Update, Delete (CRUD), kita mendefinisikan fungsi pembantu berikut untuk melakukan CRUD sebenarnya pada 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()

Fungsi tingkat db ini akan digunakan di titik akhir REST berikut:

 @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"}

Tes Tulis

Mari kita menulis beberapa tes untuk API TODO kami. FastAPI menyediakan kelas TestClient yang didasarkan pada pustaka Permintaan populer, dan kami dapat menjalankan pengujian dengan Pytest.

Untuk memastikan hanya pengguna yang masuk yang dapat membuat TODO, kita dapat menulis sesuatu seperti ini:

 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

Tes berikut memeriksa titik akhir login kami dan menghasilkan JWT jika disajikan dengan kredensial login yang valid.

 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()

Menyimpulkannya

Kami telah selesai mengimplementasikan aplikasi TODO yang sangat sederhana menggunakan FastAPI. Sekarang, Anda telah melihat kekuatan petunjuk tipe yang digunakan dengan baik dalam mendefinisikan bentuk data masuk dan keluar melalui antarmuka REST kami. Kami mendefinisikan skema di satu tempat dan menyerahkannya ke FastAPI untuk menerapkan validasi dan konversi data. Fitur penting lainnya adalah injeksi ketergantungan. Kami menggunakan konsep ini untuk mengemas logika bersama untuk mendapatkan koneksi database, mendekode JWT untuk mendapatkan pengguna yang saat ini masuk, dan menerapkan OAuth2 sederhana dengan kata sandi dan pembawa. Kami juga melihat bagaimana ketergantungan dapat dirantai bersama.

Kita dapat dengan mudah menerapkan konsep ini untuk menambahkan fitur seperti akses berbasis peran. Selain itu, kami menulis kode yang ringkas dan kuat tanpa mempelajari kekhasan kerangka kerja. Dengan kata sederhana, FastAPI adalah kumpulan alat canggih yang tidak perlu Anda pelajari karena itu hanyalah Python modern. Selamat bersenang-senang.