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란?

Python 3.6 버전 +로 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

위의 명령으로 Python 서버를 실행할 수 있습니다.

uvicorn: Python 서버를 실행하는 데 필요한 기본 명령
main : 실행할 초기 파이썬 파일 이름
app: FastAPI 모듈이 할당된 객체 이름을 의미합니다.
reload: 소스 코드가 변경될 때 서버를 다시 시작하는 옵션
port: 서버 실행 포트 옵션, 기본 8000이 기본값

지금 성공적으로 실행하면,

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로 Python의 jwt 모듈을 가져오고 encode_token이라는 메서드로 토큰을 생성합니다.

이렇게 작성한 메소드를 사용하는 read_user 를 한 번 /docs 로 실행해 봅시다.

결과는 위의 사진과 같이 표시됩니다.

위에서 나는 일시적으로 jai라는 사용자를 만들었다고 말했지만,
user_read 메소드를 통해 db에 user가 있는지 확인하면 encode_token 메소드를 통해 jwt 토큰을 생성하고 리턴하는지 확인할 수 있습니다.

없는 계정에 요청을 제출하는 경우

의 결과가 쓰러집니다.