📌 Docs 결과
축하한다!
우리는 더이상 프론트를 다루지 않아도 된다!
직접 코드를 실행함으로써 결과를 보지 않아도 된다!
fastapi로 넘어옴으로써, 우리는 docs 와 redoc을 통해 데이터가 잘 처리 되고 있는지 예시 입력들을 넣어보며 실시간으로 확인이 가능하다. 앞으로 잘 부탁해 fastapi야.
📌 비동기 작업
업무를 병렬적으로 처리하는 것을 말한다. 일반적으로 직렬적인 처리를 하지만 fastapi는 비동기적 작업처리를 지원하여, 효율적으로 코드들을 보다 빠르게 실행시킬 수 있다. 아무래도 이게 fastapi를 사용하게 되는 큰 장점이 아닌 가 싶다.
내가 작성한 응용프로그램이 다른 무엇과 소통할 필요가 없다면 async로 작성하는 것은 아마 도움이 될 것이다.
이를 사용하기 위해서는 코드에서 await를 사용한다. await는 비동기 코드에서 다른작업을 수행할 수 있는 지점을 나타낸 것이다. 동시성과 작업의 병렬성은 엄연히 다른영역이다.
내가 햄버거가게에 들어가서, 줄을 기다리고, 차례가 되었을 때, 음식을 주문하고, 또 음식이 나올 때까지 기다린다. 기다림이 굉장히 많다. 이걸 일처리라고 생각하면 된다. 특정 행동 (주문할 차례, 햄버거가 나왔을 때) 를 제외하고 다른 코드들을 실행시켜, 실행을 최적화하는 것이 비동기 작업이다.
📌 코드 이동 후기
fastapi로써의 코드적인 이동은 끝마친 상태이다. 보다 효율적이고 직관적인 코드 정리는 여전히 필요한 상황이다. 생각보다 flask와 비슷하긴 하지만, 생각보다 어려운 부분들도 많고, 무엇보다 sqlalchemy.orm을 사용하여 데이터베이스와 연결하는 부분이 아직 익숙하지 않고, 세션을 어떻게 사용해야할지도 모르겠다. 세션을 사용하는 방법을 알면 편할 것 같다. 그래서 세션에 대한 공부를 추가적으로 해야하는 상황이다
기본의 데이터 파싱보다 빠르게 하는 방법에 대해서는 아직 해답을 찾지못하여 이리저리 바꾸고 있다.
데이터를 가져와서, fastapi에서 제공해주는 Docs로 확인하는데 5초가 소요된다. 어느 부분에서 처리가 오래걸리는지, 고민해보고 바꿀 필요는 있을 것 같다.
📌 코드
search_crud
import requests
import httpx
from var import *
from models import User
from fastapi import HTTPException
from starlette import status
'''
mvc model 에서 controller 부분으로 구성
'''
api_key = Api
async def get_summoner_info(summoner_name: str):
summonerV4_url = summonerV4
url = f'{summonerV4_url}{summoner_name}?api_key={api_key}'
async with httpx.AsyncClient() as client:
response = await client.get(url)
if response.status_code == 200:
summoner_info = response.json()
return summoner_info
else:
return {"error": "Failed to retrieve summoner info"}
# ingame data는 puuid랑 match_type있으면 비동기로 실행가능
async def ingame_data(puuid, match_type):
if match_type == '솔로랭크':
match_type = 420 # 솔로랭크 엔드포인트
else:
match_type = 440 # 자유랭크 엔드포인트
match_url = f'{matchnumL}{puuid}/ids?queue={match_type}&type=ranked&start=0&count=10&api_key={api_key}'
async with httpx.AsyncClient() as client:
response = await client.get(match_url)
if response.status_code == 200:
match_id = response.json()
else:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail = "오류가 발생했습니다.",
)
match_data = []
champid = []
for m in match_id:
base_url = f'{matchL}{m}?api_key={api_key}'
r = requests.get(base_url)
Match_Dto = r.json()
MetaDataDto = Match_Dto['info']
Participants = MetaDataDto['participants']
for participant in Participants:
if participant['puuid'] == puuid:
champname = participant['championName']
champid.append(participant['championId'])
c = f'{champ_square}{champname}.png'
w = participant['win']
k = participant['kills']
d = participant['deaths']
a = participant['assists']
mdtoc = participant['magicDamageDealtToChampions']
pdtoc = participant['physicalDamageDealtToChampions']
win = 'Win' if w else 'Loss'
# zero divisionerror
try:
kda = (k + a) / d
kda = round(kda, 2)
except ZeroDivisionError:
kda = 999
match_data.append({
'champname': champname,
'champ_url': c,
'Win': win,
'Kills': k,
'Deaths': d,
'Assists': a,
'Kda': kda,
'PhysicalDtoC': pdtoc,
'MagicDtoC': mdtoc
})
return match_data
async def get_rank_data(id):
league = leagueV4
leagueV4_url = f'{league}{id}?api_key={api_key}'
response = requests.get(leagueV4_url)
summoner_rank = response.json()
for entry in summoner_rank:
rank['queue_type'] = entry["queueType"]
rank['tier'] = entry["tier"]
rank['rank'] = entry["rank"]
rank['leaguepoints'] = entry["leaguePoints"]
rank['wins'] = entry["wins"]
rank['losses'] = entry["losses"]
return rank
# 데이터 로직
def save_info(db, summoner_id, summoner_puuid, summoner_name):
db_user = User(id=summoner_id,
puuid=summoner_puuid,
name=summoner_name)
db.add(db_user)
db.commit()
def get_existing_user(db, user_name):
return db.query(User).filter(User.name == user_name).first()
search_router
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.orm import Session
from starlette import status
from database import get_db
from var import profile
from Search import search_crud, search_schema
'''
'''
router = APIRouter(
prefix="/kr"
)
riot_api = ""
@router.post("/{summoner_name}")
async def summoner_search(summoner_name: str,
_User_info: search_schema.Userinfo,
db: Session = Depends(get_db)):
summoner_name = summoner_name
match_type = "솔로랭크"
user = search_crud.get_existing_user(db, summoner_name)
if user:
print(1)
id = user.id
puuid = user.puuid
basic_info = await search_crud.get_summoner_info(summoner_name)
rank_info = await search_crud.get_rank_data(id)
match_data = await search_crud.ingame_data(puuid, match_type)
return basic_info, rank_info, match_data
else:
print(2)
basic_info = await search_crud.get_summoner_info(summoner_name)
id = basic_info['id']
puuid = basic_info['puuid']
await search_crud.save_info(id, puuid, summoner_name)
rank_info = await search_crud.get_rank_data(id)
match_data = await search_crud.ingame_data(puuid, match_type)
return basic_info, rank_info, match_data
