Python Fast Api (jwt 예제)

Python 개발자들 사이에서 점차 인기를 끌고 있는 FastAPI는, 비동기 I/O 작업을 지원하며 빠르고 쉽게 API 서버를 개발할 수 있는 프레임워크입니다. 특히, Pydantic을 이용한 데이터 유효성 검사와 자동 문서화 기능 등을 제공하여 개발 생산성을 크게 높일 수 있습니다.

이번 블로그 글에서는 FastAPI의 기본적인 개념과 사용 방법을 소개하고, JWT(JSON Web Token)를 이용한 인증 방식을 구현해보는 예제를 다룰 예정입니다. JWT는 웹 서비스에서 많이 사용되는 토큰 기반 인증 방식으로, 사용자의 정보를 안전하게 전송하고 인증하는 방식입니다.

본 글에서는 FastAPI를 이용하여 간단한 API 서버를 만들고, 해당 서버에서 JWT 토큰을 발급하고 검증하는 방법을 알아보겠습니다.

• Fast Api 란
• 기본 예시
• 부가적인 기능
• CRUD 예시
• Jwt Token 예제

  ---

1. Fast Api 란

파이썬 3.6버전+ 에서 API 서버를 구축하기 위한 웹 프레임워크 입니다.

주요 특징으로는

• 빠름 : Starlette과 Pydantic 으로 NodeJs와 대등할 정도로 높은 성능을 가지고 있습니다.
  • Starlette : 비동기적으로 실행할 수 있는 웹 어플리케이션 서버로 Uvicorn 위에서 실행
  • Uvicorn : Fast Api 를 실행하는 서버로, 초고속 ASGI Web Server입니다. 단일 프로세스에서 uvloop 기반 비동기 파이썬 코드를 실행합니다.
    • ASGI : Async Server Gateway Interface
    • uvloop : asyncio 이벤트의 루프 대체재
  • Pydantic : 타입 어노테이션을 사용해서 데이터를 검증하고 설정들을 관리하는 라이브러리
• 코드 작성이 빠름
• 버그가 적음
• 직관적
• 쉬움
• 짧음
• Swagger로 api 문서 자동화
• 별도의 구성이나 설치의 필요 없이 바로 사용 가능

CGI, WSGI, ASGI

• CGI (Common Gateway Interface) : 웹 서버에서 어플리케이션을 작동시키기 위한 인터페이스 입니다.
  • 정적으로 동작하는 웹서버를 동적으로 기능하게 만들기 위해 사용
  • 단점은 요청이 들어올때마다 프로그램을 계속해서 실행한다 -> DB Connection 을 새로 열어야 한다는 뜻
• ASGI (Asynchronous Server Gateway Interface) : WSGI 는 동기적으로만 작동하기에 여러 작업을 동시에 처리하는데 한계가 존재하기에, 비동기로 처리할 수 있는 ASGI 가 등장했습니다.
  • 대표적인 친구가 Uvicorn 입니다.

Fast API 설치

# fastapi 설치
pip3 install fastapi

# uvicorn 설치 (순수 파이썬)
pip3 install uvicorn

# uvicorn 설치 (C / C++도 섞인 애)
pip3 install 'uvicorn[standard]'

---

기본 예시

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/") # 라우트 경로
def test() :
    return {
        "python" : "Framework"
    }

이런식으로 작성한 뒤 해당 파일이 있는 폴더에서

uvicorn main:app --relaod --port 8000

위 명령어를 통해 파이썬 서버를 실행할 수 있습니다.

• uvicorn : 파이썬 서버 실행을 위해 필요한 기본 명령어
• main : 실행할 초기 파이썬 파일 이름
• app: FastAPI 모듈을 할당한 객체명을 뜻함
• reload : 소스코드가 변경되었을 때, 서버 재시작 옵션
• port : 서버 실행 포트 옵션, 생략시 8000이 default

이제 정상적으로 실행을 하였을 때,

INFO:     Will watch for changes in these directories: ['C:\\dev\\openobject\\study\\python\\test']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1752] using StatReload
INFO:     Started server process [2128]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

위의 내용을 터미널에서 확인할 수 있습니다.

그리고 나와있는 포트로 접속을 하게 되면

이러한 결과값을 볼 수 있습니다.

---

부가적인 기능

Fast API는 작성된 API들을 웹 상에서 바로 테스트할 수 있는 기능을 Swagger Api 로 제공을 합니다. 접속 방법은 실행된 상태에서 /docs url을 접속하여 확인할 수 있습니다.

접속 하면

위와 같은 화면을 확인할 수 있습니다.

---

CRUD 예시

fastapi 에서 사용하는 ORM 으로는 sqlalchemy를 사용합니다.

먼저 DB 를 연결해야 합니다.

# DB.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

user_name = "root"
user_pwd = "DB 비밀번호" #본인 db 비밀번호로 변경
db_host = "127.0.0.1"
db_name = "crudpy" #임의로 설정했음

DATABASE =  'mysql://%s:%s@%s/%s?charset=utf8' % (
    user_name,
    user_pwd,
    db_host,
    db_name
)

ENGINE = create_engine(
    DATABASE,
    encoding="utf-8",
    echo=True
) # 위의 정보들을 담은 상태에서 utf-8의 인코딩 방식으을 가진 DB 접속 엔진을 생성합니다.

session = scoped_session(
    sessionmaker(
        autocommit = False,
        autoflush = False,
        bind = ENGINE
    )
)

Base = declarative_base()
Base.query = session.query_property()

그리고 이름과 나이를 가진 모델을 한번 만들어보겠습니당

# model.py
from sqlalchemy import Column, Integer, String
from pydantic import BaseModel
from db import Base
from db import ENGINE

class UserTable(Base): # 유저 테이블
    __tablename__ = 'user' #테이블 이름
    id = Column(Integer, primary_key=True, autoincrement=True) # 숫자 형식의 index 번호 자동 생성
    user_name = Column(String(50), nullable=False) # 스트링 형식의 널 삽입 불가 컬럼
    age = Column(Integer) # 숫자 컬럼

class User(BaseModel): # 유저
    id: int
    user_name: str
    age: int

자 이렇게 DB 와 모델을 만들었으면 적용을 해봐야죠!

위의 내용들을 연결해줄 main.py를 만들어보겠습니다.

# main.py
from fastapi import FastAPI
from typing import List
from starlette.middleware.cors import CORSMiddleware

from db import session
from model import UserTable, User

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credential=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

@app.get("/users")
async def read_users(): # 유저 리스트 가져오기
    users = session.query(UserTable).all()
    return users

@app.get("/users/{user_id}")
async def read_user(user_id: int): # 특정 유저 가져오기
    user = session.query(UserTable).filter(UserTable.id == user_id).first()
    return user

@app.post("/users")
async def create_user(name: str, age: int) : # 유저 생성
    user = UserTable()
    user.name = name
    user.age = age

    session.add(user)
    session.commit()

    return f"{name} created..."

@app.put("/users")
async def update_user(users: List[User]): # 유저 업데이트
    for i in users:
        user = session.query(UserTable).filter(UserTable.id == i.id).first()
        user.name = i.name
        user.age = i.age
        session.commit()
    return f"{users[0].name} updated..."

@app.delete("/users")
async def delete_users(user_id: int): # 유저 삭제
    user = session.query(UserTable).filter(UserTable.id == user_id).delete()
    session.commit()

    return read_usres

이런식으로 CRUD 예제를 만들어보았습니다.

이렇게 작성된 main.py를 /docs 에 들어가서 확인하게 된다면

위와 같이 나오게 됩니다.

한번 여기서 유저 리스트를 확인해보면 아래와 같이 제가 임시로 넣어둔 유저 정보가 나오게 됩니다.

---

JWT 토큰 예시

자 이제 API 에서 가장 자주 사용되는 기능인 로그인에서 사용되는 JWT 토큰을 적용해보는 예제를 작성해보겠습니다.

위에서 작성된 문서에서 살짝 변형을 하여

from logging import error
import jwt
from datetime import timedelta, datetime
from fastapi import Depends, FastAPI
from typing import List
from starlette.middleware.cors import CORSMiddleware

from db import session
from model import UserTable, User
from fastapi.security import OAuth2PasswordBearer

# OAuth2PasswordBearer 객체를 생성할 때 tokenUrl이라는 파라미터를 넘겨준다. 프론트엔드에서 token 값을 얻어 올 때 사용한다.
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

@app.get("/users")
async def read_users():
    users = session.query(UserTable).all()
    return users

def encode_token(username):
    if username :
        payload = {
            'exp': datetime.utcnow() + timedelta(weeks=5),
            'iat': datetime.utcnow(),
            'scope': 'access_token',
            'data': username
        }
        return jwt.encode(
            payload,
            'test',
            algorithm='HS256'
        )
    else :
        return error

@app.get("/users/{user_name}")
async def read_user(user_name):
    user = session.query(UserTable).filter(UserTable.user_name == user_name).first()
    if user :
        return encode_token(user.user_name)
    else :
        return user

@app.post("/user")
async def create_users(user_name: str, age: int):
    user = UserTable()
    user.user_name = user_name
    user.age = age

    session.add(user)
    session.commit()

    return f"{user_name} created"

@app.put("/users")
async def update_user(users: List[User]):
    for i in users:
        user = session.query(UserTable).filter(UserTable.id == i.id).first()
        user.user_name = i.user_name
        user.age = i.age
        session.commit()

    return f"{users[0].user_name} updated"

@app.delete("/user")
async def delete_users(user_id: int):
    user = session.query(UserTable).filter(UserTable.id == user_id).delete()
    session.commit()

    return read_users

이런식으로 작성할 수 있습니다.

여기서 jwt 토큰을 생성하는 부분은

import jwt
...
# OAuth2PasswordBearer 객체를 생성할 때 tokenUrl이라는 파라미터를 넘겨준다. 프론트엔드에서 token 값을 얻어 올 때 사용한다.
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

...
def encode_token(username):
    if username :
        payload = {
            'exp': datetime.utcnow() + timedelta(weeks=5),
            'iat': datetime.utcnow(),
            'scope': 'access_token',
            'data': username
        }
        return jwt.encode(
            payload,
            'test',
            algorithm='HS256'
        )
    else :
        return error

이 부분으로 oauth2_scheme는 만들어두고 사용하는 부분이 없어 다음에 설명하겠습니다.

먼저 import jwt 로 파이썬 내의 jwt 모듈을 가져오고, encode_token이라는 메서드에서 토큰을 생성해줍니다.

이렇게 만든 메서드를 사용하는 read_user를 한번 /docs 에서 실행해보겠습니다.

결과는 위의 사진과 같이 나옵니다. 위에서 제가 임시로 jai라는 user를 만들었다고 했었는데,
user_read 메서드를 통해 db에 user가 있는지 확인 후 있다면 encode_token 메서드를 통해 jwt토큰을 생성 리턴을 하는 것으로 확인해볼 수 있습니다.

만약 없는 계정에 대해서 요청을 보내게 된다면

의 결과가 넘어오게 됩니다.