Python을 사용한 고성능 앱 – FastAPI 자습서

게시 됨: 2022-03-11

좋은 프로그래밍 언어 프레임워크를 사용하면 고품질 제품을 더 빨리 생산할 수 있습니다. 훌륭한 프레임워크는 전체 개발 경험을 즐겁게 만듭니다. FastAPI는 강력하고 즐겁게 사용할 수 있는 새로운 Python 웹 프레임워크입니다. 다음 기능을 사용하면 FastAPI를 시도해 볼 가치가 있습니다.

  • 속도: FastAPI는 가장 빠른 Python 웹 프레임워크 중 하나입니다. 실제로 속도는 Node.js 및 Go와 동등합니다. 이러한 성능 테스트를 확인하십시오.
  • 상세하고 사용하기 쉬운 개발자 문서
  • 코드 힌트를 입력하고 무료 데이터 유효성 검사 및 변환을 받으세요.
  • 의존성 주입을 사용하여 쉽게 플러그인을 만듭니다.

TODO 앱 구축

FastAPI 이면의 큰 아이디어를 탐색하기 위해 사용자를 위한 할 일 목록을 설정하는 TODO 앱을 빌드해 보겠습니다. 우리의 작은 앱은 다음 기능을 제공합니다:

  • 가입 및 로그인
  • 새 TODO 항목 추가
  • 모든 TODO 목록 가져오기
  • TODO 항목 삭제/업데이트

데이터 모델을 위한 SQLAlchemy

우리 앱에는 사용자와 TODO라는 두 가지 모델만 있습니다. Python용 데이터베이스 툴킷인 SQLAlchemy의 도움으로 모델을 다음과 같이 표현할 수 있습니다.

 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 정보가 들어오고 나가는 모든 방식에 대해 생각하십시오.

전통적으로 새로운 사용자는 TODO 서비스에 가입하고 기존 사용자는 로그인합니다. 이 두 상호 작용은 모두 사용자 정보를 다루지만 데이터의 형태가 다릅니다. 가입하는 동안 사용자로부터 더 많은 정보가 필요하고 로그인할 때는 최소한의 정보(이메일과 비밀번호만)가 필요합니다. 즉, 이 두 가지 다른 형태의 사용자 정보를 표현하려면 두 가지 pydantic 모델이 필요합니다.

그러나 TODO 앱에서는 JSON 웹 토큰(JWT) 기반 로그인 흐름을 위해 FastAPI에 내장된 OAuth2 지원을 활용할 것입니다. 등록 끝점으로 흘러갈 데이터를 지정하기 위해 여기에서 UserCreate 스키마를 정의하고 등록 프로세스가 성공한 경우 응답으로 반환할 UserBase 스키마를 정의하기만 하면 됩니다.

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

여기에서 성, 이름 및 비밀번호를 문자열로 표시했지만 최소 길이, 최대 길이 및 정규식과 같은 검사를 가능하게 하는 pydantic 제한 문자열을 사용하여 더 강화할 수 있습니다.

TODO 항목의 생성 및 나열을 지원하기 위해 다음 스키마를 정의합니다.

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

기존 TODO 항목의 업데이트를 지원하기 위해 다른 스키마를 정의합니다.

 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 웹 토큰을 발급합니다.

자격 증명을 받기 위해 FastAPI 보안 유틸리티의 일부인 OAuth2PasswordRequestForm 을 사용합니다.

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

종속성 주입을 사용하여 DB 액세스 및 엔드포인트 보호

로그인 성공 시 사용자에게 JWT를 제공하는 로그인 끝점을 설정했습니다. 사용자는 이 토큰을 로컬 저장소에 저장하고 백엔드에 Authorization 헤더로 표시할 수 있습니다. 로그인한 사용자의 액세스만 예상하는 엔드포인트는 토큰을 디코딩하고 요청자가 누구인지 알아낼 수 있습니다. 이러한 종류의 작업은 특정 끝점에 연결되지 않고 모든 보호된 끝점에서 사용되는 공유 논리입니다. 모든 요청 처리기에서 사용할 수 있는 종속성으로 토큰 디코딩 논리를 설정하는 것이 가장 좋습니다.

FastAPI로 말하면 경로 작업 기능(요청 처리기)은 get_current_user 에 의존합니다. get_current_user 종속성은 데이터베이스에 연결하고 FastAPI의 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

로그인한 사용자는 TODO를 CRUD할 수 있습니다.

TODO 생성, 읽기, 업데이트, 삭제(CRUD)에 대한 경로 연산 함수를 작성하기 전에 db에서 실제 CRUD를 수행하기 위해 다음 도우미 함수를 정의합니다.

 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는 인기 있는 Requests 라이브러리를 기반으로 하는 TestClient 클래스를 제공하며 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()

요약

FastAPI를 사용하여 매우 간단한 TODO 앱 구현을 완료했습니다. 지금까지 REST 인터페이스를 통해 들어오고 나가는 데이터의 모양을 정의하는 데 유용하게 사용되는 유형 힌트의 힘을 보았습니다. 스키마를 한 곳에서 정의하고 데이터 유효성 검사 및 변환을 적용하기 위해 FastAPI에 맡깁니다. 다른 주목할만한 기능은 종속성 주입입니다. 우리는 이 개념을 사용하여 데이터베이스 연결을 얻고, JWT를 디코딩하여 현재 로그인한 사용자를 가져오고, 비밀번호와 베어러를 사용하여 간단한 OAuth2를 구현하는 공유 로직을 패키징했습니다. 또한 종속성을 함께 연결할 수 있는 방법도 보았습니다.

이 개념을 쉽게 적용하여 역할 기반 액세스와 같은 기능을 추가할 수 있습니다. 게다가 우리는 프레임워크의 특성을 배우지 않고 간결하고 강력한 코드를 작성합니다. 간단히 말해서 FastAPI는 최신 Python이기 때문에 배울 필요가 없는 강력한 도구 모음입니다. 즐거운 시간 보내세요.