fastapi-ex1

Published on 21-10-2721-10-27

https://www.youtube.com/watch?v=0CQdCfICAPA

conda install requests

conda install -c conda-forge python-multipart

FastAPI 기초 (1/5) – 개요 및 설치하기

Swagger : http://127.0.0.1:8000/docs
ReDoc: http://127.0.0.1:8000/redoc
openapi: http://127.0.0.1:8000/openapi.json

http://globalintelligence.kr/ >> GNC활동관리 >> Datalab

FastAPI 기초 (2/5) – API구현 (GET, POST, DELETE)

: RESTful을 fastAPI로 구현

예제> Intro to FastAPI – The Best Way to Create APIs in Python?

https://worldtimeapi.org/timezones

import uvicorn
from fastapi import FastAPI

from app.routes import index #auth

def create_app():
    app=FastAPI()
    app.include_router(index.router)
    return app

app = create_app()

# uvicorn main:app --reload
if __name__ == "__main__" :
    uvicorn.run("main:app", port=8888, reload=True)

from fastapi import APIRouter
from fastapi import File, UploadFile

import requests
from starlette.requests import Request
from starlette.responses import JSONResponse

from pydantic import BaseModel
import json

#from inspect import currentframe as frame
from datetime import datetime
from dataclasses import asdict
from typing import Optional

router = APIRouter()

db = []

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

@router.get("/")
async def root(req: Request):
    """
    `ELB 상태 체크용 API` \

    :return:
    """
    return JSONResponse( {'hello':'sixxx'} )

@router.get("/cities")
def get_cities():
    results = []
    for aCity in db:
        strs = f"http://worldtimeapi.org/api/timezone/{aCity['timezone']}"
        r = requests.get(strs)
        cur_time = r.json()['datetime']
        results.append(
            {'name':aCity['name'],
                'timezone':aCity['timezone'],
                'current_time':cur_time}
        )
    return results

...

@router.delete("/cities/{city_id}")
def delete_city(city_id:int):
    db.pop(city_id-1) 
    return {}

FastAPI 기초 (3/5) – 웹CRUD ( GET, POST, PUT, DELETE )

FastAPI 기초 (4/5) – MySQL (MariaDB) 연동 CRUD

Starlette : 경량 ASGI 프레임웤

CREATE TABLE `user` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(50) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  `age` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

import uvicorn
from fastapi import FastAPI
from app.routes import dbdb

def create_app():
    app=FastAPI()
    app.include_router(dbdb.router)    
    return app

app = create_app()

from starlette.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)
# uvicorn main:app --reload
if __name__ == "__main__" :
    uvicorn.run("main:app", port=8888, reload=True)

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

user_name = 'enduser'
user_pwd  = '1234'
db_host   = 'localhost'
db_name   = 'edudb'

DATABASE = "mysql+pymysql://%s:%s@%s/%s?charset=utf8mb4" %(
    user_name, user_pwd, db_host, db_name)
#DATABASE = "mysql+pymysql://enduser:1234@localhost/edudb?charset=utf8mb4"

ENGINE = create_engine(
    DATABASE,
    encoding="utf-8",
    echo=True
)

session = scoped_session(
    sessionmaker(
        autocommit=False, 
        autoflush=False,
        bind = ENGINE
    )
)
# create instance
Base = declarative_base()
Base.query = session.query_property()

from sqlalchemy import Column, Integer, String
from pydantic import BaseModel

from database.conn import Base
from database.conn import ENGINE

class UserTable(Base):
    __tablename__ = 'user'
    id  = Column(Integer, primary_key=True,  autoincrement=True)
    name= Column(String(50)) # , nullable=False)
    age = Column(Integer)

class User(BaseModel):
    id : int
    name : str
    age : int

# SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age 
# FROM user)

from fastapi import APIRouter 
from typing import List

import sys, os
sys.path.append(os.path.dirname(os.path.abspath(os.path.dirname(__file__))))
#sys.path.append('/Users/onesixx/my/git/hello/app')

from database.conn  import session
from database.model import UserTable, User

import requests
from starlette.requests import Request
from starlette.responses import JSONResponse

router = APIRouter()

@router.get("/")
def read_users():
    return JSONResponse( {'hello':'sixxx'} )

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

@router.get("/users/{user_id}")
def read_users(user_id: int):
    user= session.query(UserTable).filter(UserTable.id==user_id).first()
    return user

@router.post("/users")
def create_users(name:str, age:int):
    user= UserTable()
    user.name = name
    user.age = age
    session.add(user)
    session.commit()
    return f"{name} created."

@router.put("/users")
def update_users(userList: List[User]):
    for i in userList:
        user = session.query(UserTable).filter(UserTable.id == i.id).first()
        user.name= i.name
        user.age = i.age
        session.commit()

    return f"{userList[0].name} updated..."

@router.delete("/users/{user_id}")
def delete_users(user_id: int):
    user = session.query(UserTable).filter(UserTable.id == user_id).delete()
    session.commit()
    return  "delete..."