FastAPI로 CRUD, ORM sqlalchemy 사용, 캐쉬 예시 예제

FastAPI로 CRUD, ORM sqlalchemy 사용, 캐쉬 예시 예제

 

 

FastAPI로 CRUD

모듈 설치

pip install fastapi
pip install "uvicorn[standard]"
pip install requests
pip install pydantic

 

코드 작성

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import requests

app = FastAPI()

# this is list now, but it's gonna real database like mongoDB
db = []


# class must be pydantic field type to POST in FastAPI
class City(BaseModel):
    name: Optional[str] = None
    timezone: Optional[str] = None


# main http
@app.get("/", description="This is World Time CRUD API", response_model=str)
def root():
    return {"hello this is World Time CRUD API!!"}


# get all city's info in db and extract time with using <http://worldtimeapi.org>
# if you put optional parameter, you can get data through process of handling parameter
@app.get('/cities', description="You can get cities data")
def get_cities(name: Optional[str] = None):
    results = []
    if name:
        for city in db:
            if city["name"] == name:
                strs = f"http://worldtimeapi.org/api/timezone/{city['timezone']}"
                r = requests.get(strs)
                cur_time = r.json()['datetime']
                results.append({'name': city['name'], 'timezone': city['timezone'], 'current_time': cur_time})

    else:
        for city in db:
            strs = f"http://worldtimeapi.org/api/timezone/{city['timezone']}"
            r = requests.get(strs)
            cur_time = r.json()['datetime']
            results.append({'name': city['name'], 'timezone': city['timezone'], 'current_time': cur_time})

    return results


# get one searched city's info in db and extract time with using <http://worldtimeapi.org>
@app.get('/cities/{city_id}', description="You can get city data")
def get_city(city_id: int):
    if len(db) < city_id:
        raise HTTPException(status_code=404, detail="Item not found")

    city = db[city_id - 1]
    strs = f"http://worldtimeapi.org/api/timezone/{city['timezone']}"
    r = requests.get(strs)
    cur_time = r.json()['datetime']

    return {'name': city['name'], 'timezone': city['timezone'], 'current_time': cur_time}


# create one city data and post it to db
@app.post('/cities', description="You can create city data")
def create_city(city: City):
    db.append(city.dict())

    return db[-1]


# delete one data in db.
@app.delete('/cities/{city_id}', description="You can delete city data, it will return result", response_model=bool)
def delete_city(city_id: int):
    if len(db) < city_id:
        isdeleted = False
    else:
        db.pop(city_id - 1)
        isdeleted = True

    return isdeleted


# update all attribute from one data in db.
@app.put('/cities/{city_id}', description="You can update city data")
def update_city(city_id: int, city: City):
    db[city_id - 1] = city.dict()

    return {}


# update some attributes at one data in db.
@app.patch('/cities/{city_id}', description="You can update some attribute at one data")
def update_city(city_id: int, city: City):
    db[city_id - 1]['name'] = city.name;

    return {}

 

 

실행

uvicorn main:app --reload

 

 

확인

Fast API는 Swagger UI, Redoc을 통해 Automatic Docs를 생성하는것을 지원한다.

http://127.0.0.1:8000/docs - SwaggerUI

http://127.0.0.1:8000/redoc - Redoc

 

Swagger UI에서는 브라우저에서 API TEST 기능을 수행 해 볼 수 있다.

 

 

 

ORM ( SQLALCHEMY 이용)

• mysql 설치

MySQL :: MySQL Community Downloads

 

MySQL Command Line Client 실행

CREATE DATABASE city;
USE city;
create table city (
    name VARCHAR(30) NOT NULL,
    timezone VARCHAR(30) NOT NULL);

 

 

• fastapi-sqlalchemy, pymysql 설치

$ pip3 install fastapi-sqlalchemy 
$ pip3 install pymysql

 

• 모델 생성 (models.py)

from sqlalchemy import Column, String
from sqlalchemy.orm import declarative_base


Base = declarative_base()

class City(Base):
  __tablename__ = 'city'
  name = Column(String, primary_key=True)
  timezone = Column(String, nullable=False)

 

 

 메인 서버 변경 (main.py) - CRUD

from fastapi import FastAPI
from fastapi_sqlalchemy import DBSessionMiddleware
from fastapi_sqlalchemy import db
from pydantic import BaseModel

from models import City



class Item(BaseModel):
    name : str
    timezone : str


HOSTNAME = 'localhost'
PORT = 3306
USERNAME = 'root'
PASSWORD = ''
DBNAME = 'city'
MYSQL_URL = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOSTNAME}:{PORT}/{DBNAME}'

app = FastAPI()

app.add_middleware(DBSessionMiddleware, db_url=MYSQL_URL)

@app.get("/")
def root():
    return {"hello this is World Time CRUD API!!"}

@app.get('/cities')
def get_cities():
  query = db.session.query(City)
  return query.all()

@app.get('/cities/{city_name}')
def get_city(city_name: str):
    query = db.session.query(City).filter(City.name == city_name)
    return query.all()

@app.post('/cities')
def create_city(city : Item):
    ed_city = City(name=city.name, timezone = city.timezone)
    db.session.add(ed_city)
    db.session.commit()
    return

@app.delete('/cities/{city_name}')
def delete_city(city_name: str):
    db.session.query(City).filter(City.name == city_name).delete()
    return

 

 

- 확인

 

 

 

 

캐쉬 적용

from fastapi import FastAPI
from fastapi_sqlalchemy import DBSessionMiddleware
from fastapi_sqlalchemy import db
from pydantic import BaseModel
from models import City



class Item(BaseModel):
    name : str
    timezone : str


HOSTNAME = 'localhost'
PORT = 3306
USERNAME = 'root'
PASSWORD = ''
DBNAME = 'city'
MYSQL_URL = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOSTNAME}:{PORT}/{DBNAME}'

app = FastAPI()

app.add_middleware(DBSessionMiddleware, db_url=MYSQL_URL)

cache = {}

@app.get("/")
def root():
    return {"hello this is World Time CRUD API!!"}

@app.get('/cities')
def get_cities():
  query = db.session.query(City)
  return query.all()

@app.get('/cities/{city_name}')
def get_city(city_name: str):
    if cache[city_name]:
        return {'name': city_name, 'timezone': cache[city_name]}
    
    query = db.session.query(City).filter(City.name == city_name)
    return query.all()

@app.post('/cities')
def create_city(city : Item):
    ed_city = City(name=city.name, timezone = city.timezone)
    db.session.add(ed_city)
    db.session.commit()
    
    cache[city.name] = city.timezone
    return

@app.delete('/cities/{city_name}')
def delete_city(city_name: str):
    db.session.query(City).filter(City.name == city_name).delete()
    return

 

보통 캐쉬로는 Redis를 많이 사용한다.