DB 연동 Fast API

리눅스 WSL(윈도우 서브시스템 리눅스)에 DB설치

WSL 가상환경에 Mysql 설치

  • sudo apt_get update
  • sudo apt-get install mysql-server
  • sudo vi/etc/mysql/mysql.conf.d/mysqld.cnf
    • 여기서 에디터 상태로 바꿔서, bind-address = 0.0.0.0 으로 수정한다. (모든 ip 접속 가능하게 함) 

Mysql에 계정 추가, 권한 추가, Mysql 재시작


Workbench(MySQL 에디터)로 접속 가능한지 체크


DB 연결을 위한 라이브러리 설치

pip install SQLAlchemy==1.4.0 설치
pip install mysqlclient==2.0.0 설치


실습1

main.py

# fastapi 라이브러리에서 FastAPI 클래스를 가져옴
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from typing_extensions import Annotated
import uvicorn
from sqlalchemy import create_engine
# from fastapi import Form

db_connection = create_engine('mysql+pymysql://test:1234@127.0.0.1:3306/test')

templates = Jinja2Templates(directory="templates")
app = FastAPI()


@app.get("/")
def hello():
  return {"message":"안녕하세요 fastAPI입니다."}

@app.get("/mysqltest")
def mysqltest(request: Request):
  query = db_connection.execute("select * from player")
  result_db = query.fetchall()

  # 결과 담을 그릇
  result = []
  for data in result_db:
    temp = {'player_id':data[0], 'player_name':data[1]}
    result.append(temp)
   
  return templates.TemplateResponse("sqltest.html", {'request':request, 'result_table':result})



# detail 출력
@app.get("/detail")
def test_post(request: Request, id: str, name: str):
    print(id, name)
    # SQL Injection 방지를 위해 바인딩 처리
    query = db_connection.execute("SELECT * FROM player WHERE player_id = {} AND player_name LIKE '{}'".format(id,name))
    result_db = query.fetchall()
    result = []
    for i in result_db:
        temp = {
            'player_id': i[0],
            'player_name': i[1],
            'team_name': i[2],
            'height': i[-2],
            'weight': i[-1]
        }
        result.append(temp)
    return templates.TemplateResponse("detail.html", {"request": request, "result_table": result})


# 웹주소 접속시 자동 reload
if __name__ == '__main__':
  uvicorn.run(app, host="localhost", port=8000)


sqltest.html > 버튼 클릭 > main.py > detail.html

<button onClick="location.href='/detail?id={{row.player_id}}&name={{row.player_name}}'">상세정보</button>

sqltest.html

<html>
  <head>
    <style>
      table, th, td{
        border: 1px solid black;
      }
    </style>
    <meta charset="UTF-8">
    <title>회원관리</title>
  </head>
  <body>
    <table>
      <tr>
        <th>선수id</th>
        <th>선수이름</th>
      </tr>
     
      {% for row in result_table%}
      <tr>
        <td>{{row.player_id}}</td>
        <td>{{row.player_name}}</td>
        <td><button onClick="location.href='/detail?id={{row.player_id}}&name={{row.player_name}}'">상세정보</button></td>
      </tr>
      {% endfor %}
    </table>
  </body>
</html>


detail.html

<!DOCTYPE html>
<html>
<head>
<style>
table, th, td, tr{
    border : 1px solid red;
}
</style>
<meta charset="UTF-8">
<title>회원관리</title>
</head>
<body>
    <h2> 선수 상세정보 </h2>
<table>
    <tr>
        <th>선수id</th>
        <th>선수이름</th>
        <th>팀이름</th>
        <th>weight</th>
        <th>height</th>
    </tr>
        {% for row in result_table%}
        <tr>
            <td>{{row.player_id}}</td>
            <td>{{row.player_name}}</td>
            <td>{{row.team_name}}</td>
            <td>{{row.weight}}</td>
            <td>{{row.height}}</td>
        </tr>
        {% endfor %}
</table>
</body>
</html>



실습2

main3.py
# 회원가입 페이지
@app.get("/regi")
def regi(request: Request):
  return templates.TemplateResponse("regi.html", {'request': request})


# 회원가입
@app.post("/insert")
def post_insert(request: Request, name: Annotated[str, Form()], id: Annotated[str, Form()], pw: Annotated[str, Form()], email: Annotated[str, Form()]):
   
    query = f"insert into student(sname, s_id, s_pwd, email) \
              values('{name}', '{id}', '{pw}', '{email}');"

    db_connection.execute(query)

    return templates.TemplateResponse("login.html", {'request': request})



# 로그인 페이지
@app.get("/login")
def regi(request: Request):
  return templates.TemplateResponse("login.html", {'request': request})


# 로그인
@app.post("/login_check")
def post_login_check(request: Request, id: Annotated[str, Form()], pw: Annotated[str, Form()]):
   
    query = f"select sname from student \
              where s_id = '{id}' and s_pwd = '{pw}';"

    result_db = db_connection.execute(query)
    result_all = result_db.fetchall()

    if len(result_all) == 0:
       return {"message":"회원이 존재하지 않습니다."}
    else:
       return {"message":f"{result_all[0][0]}님 환영합니다."}


# 웹주소 접속시 자동 reload
if __name__ == '__main__':
  uvicorn.run(app, host="localhost", port=8000)


regi.html
  <form action="/insert" method="post">
    <h1>학생 등록 페이지</h1>
    <p>이름: <input type="text" name="name"></p>
    <p>아이디: <input type="text" name="id"></p>
    <p>비밀번호: <input type="password" name="pw"></p>
    <p>이메일: <input type="text" name="email"></p>
    <p><input type="reset" value="취소">&nbsp;&nbsp;<input type="submit" value="가입"></p>
  </form>


login.html
  <form action="/login_check" method="post">
    <h1>로그인 페이지</h1>
    <p>아이디: <input type="text" name="id"></p>
    <p>비밀번호: <input type="password" name="pw"></p>
    <p><input type="submit" value="로그인"></p>
  </form>





댓글 쓰기

다음 이전