251029 [ Day 76 ] - SQL (5)

시작하며

오늘은 다시 MySQL에 대해 배웠다. Python에서 MySQL을 사용하는 것부터 FastAPI를 활용해 MySQL과 FastAPI를 기본적으로 연동하는 것까지 배워보았다.

mysql-connector 설치

• PIP
  • pip install mysql-connector-python
• Conda
  • conda install anaconda::mysql-connector-python

Python에서 실행

• cursor : DB 서버와 Python 사이에서 SQL문을 실행하고, 그 결과를 가져오는 중간 관리자
• fetchall() : 결과 전체를 리스트로 반환
• fetchone() : 결과에서 1개 행만 반환
• execute() : SQL 쿼리를 실행하고, DB에 명령을 전달
• commit() : 현재 세션에 있는 모든 변경사항을 DB에 반영

# MySQL 라이브러리 불러오기
import mysql.connector

# MySQL 연결정보로 연결
conn = mysql.connector.connect(
    host="localhost",
    user="user", # 지정한 유저 이름
    password="1234", # 지정한 비밀번호
    database="codingon_db" # 지정한 데이터베이스
)

# SQL 실행을 위한 cursor 객체 생성
cursor = conn.cursor()

# 쿼리 실행
cursor.execute("SELECT * FROM users")

# 결과 출력
for row in cursor.fetchall():
    print(row)

# 연결 종료
conn.close()

• 예외 처리 적용

try:
    # 실행 코드
    conn = mysql.connector.connect(
        host="localhost",
        user="user", # 지정한 유저 이름
        password="1234", # 지정한 비밀번호
        database="codingon_db" # 지정한 데이터베이스
    )
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    for row in cursor.fetchall():
        print(row)
# 에러 처리
except Error as e:
    print("DB 연결 중 에러 발생 :", e)
finally:
    if conn.is_connected():
        # 연결 종료
        cursor.close()
        conn.close()
        print("MySQL Connection is Closed!")

CRUD

• MySQL 쿼리에서는 플레이스 홀더(%s) 사용

sql = "INSERT INTO users (username, password, name, email) VALUES (%s, %s, %s, %s)"
user_data = (username, passwoed, name, email)

cursor.execute(sql, user_data)

CREATE (INSERT)

try:
    conn = mysql.connector.connect(
        host="localhost",
        user="user",
        password="1234",
        database="codingon_db"
    )
    cursor = conn.cursor()

    # 사용자 정보 입력 받기
    username = input("사용자 ID를 입력하세요 : ")
    passwoed = input("비밀번호를 입력하세요 : ")
    name = input("이름을 입력하세요 : ")
    email = input("이메일을 입력하세요 : ")

    # INSERT문 사용하기
    sql = "INSERT INTO users (username, password, name, email) VALUES (%s, %s, %s, %s)"
    user_data = (username, passwoed, name, email)

    # SQL 실행
    cursor.execute(sql, user_data)

    # 실제 DB에 반영
    conn.commit()
    print("사용자 등록이 완료되었습니다!")

    # 쿼리 실행
    cursor.execute("SELECT * FROM users")

    # 결과 출력
    for row in cursor.fetchall():
        print(row)
except Error as e:
    print("DB 연결 중 에러 발생 :", e)
finally:
    if conn.is_connected():
        cursor.close()
        conn.close()
        print("MySQL Connection is Closed!")

READ (SELECT)

try:
    conn = mysql.connector.connect(
        host="localhost",
        user="user",
        password="1234",
        database="codingon_db"
    )
    cursor = conn.cursor()

    # SELECT문 사용하기
    # 사용자 입력 받기
    username = input("ID : ")
    passwoed = input("PW : ")

		# SQL문
    sql = "SELECT * FROM users WHERE username = %s AND password = %s"
    user_data = (username, passwoed)

		# SQL 실행
    cursor.execute(sql, user_data)
    user = cursor.fetchone()

    if user:
        print(f"로그인 성공! 환영합니다, {user[3]}님")
        print("사용자 정보 :", user)
    else:
        print("로그인 실패")

except Error as e:
    print("DB 연결 중 에러 발생 :", e)
finally:
    if conn.is_connected():
        cursor.close()
        conn.close()
        print("MySQL Connection is Closed!")

UPDATE (UPDATE)

try:
    conn = mysql.connector.connect(
        host="localhost",
        user="user",
        password="1234",
        database="codingon_db"
    )
    cursor = conn.cursor()

    # UPDATE문 사용하기
    user_id = input("수정할 사용자 ID : ")
    new_name = input("새 이름 : ")
    new_email = input("새 이메일 : ")

    # SQL문
    sql = "UPDATE users SET name = %s, email = %s WHERE user_id = %s"
    new_data = (new_name, new_email, user_id)

    # SQL 실행
    cursor.execute(sql, new_data)

    # DB 저장
    conn.commit()
    print("사용자 정보가 수정되었습니다!")

except Error as e:
    print("DB 연결 중 에러 발생 :", e)
finally:
    if conn.is_connected():
        cursor.close()
        conn.close()
        print("MySQL Connection is Closed!")

DELETE (DELETE)

try:
    conn = mysql.connector.connect(
        host="localhost",
        user="user",
        password="1234",
        database="codingon_db"
    )
    cursor = conn.cursor()

    # DELETE문 사용하기
    user_id = input("삭제할 사용자 ID : ")

    # SQL문
    sql = "DELETE FROM users WHERE user_id = %s"

    # SQL 실행
    cursor.execute(sql, (user_id,))

    # DB 저장
    conn.commit()
    print("사용자가 삭제되었습니다!")

except Error as e:
    print("DB 연결 중 에러 발생 :", e)
finally:
    if conn.is_connected():
        cursor.close()
        conn.close()
        print("MySQL Connection is Closed!")

Restful API

• REST : REpresentational State Transfer의 약자
  • 자원을 HTTP URI로 표현
  • HTTP 메서드를 이용해 자원을 조작하는 방식의 API 설계 원칙
• RESTful API : REST 원칙을 지키며 설계된 API를 의미

구성 요소

• 자원(Resource) : 무엇을 다루는지
• 행위(Method) : 무엇을 할 건지
• 표현(Representation) : 자원을 어떻게 주고받을지

메서드와 의미

• GET : 데이터 조회
  • GET /users : 사용자 목록 조회
  • GET /users/1 : 특정 사용자 정보 조회
• POST : 새 데이터 생성
  • POST /users : 새 사용자 생성
• PUT : 데이터 전체 수정
  • PUT /users/1 : 사용자 정보 전체 수정
• PATCH : 데이터 일부 수정
• DELETE : 데이터 삭제
  • DELETE /users/1 : 사용자 삭제

FastAPI와 함께 사용

• Python 웹 프레임워크
  • django
  • Flask
  • FastAPI

FastAPI

• Python 기반의 고성능 웹 프레임워크
• FastAPI 설치 : pip install "fastapi[standard]”
• 특징
  • RESTful API 서버 구축에 최적화
  • 비동기(async/await) 완벽 지원 → 빠른 처리 속도
  • Pydantic 기반 타입 검증 → 자동 데이터 유효성 검사
  • 자동 문서화 기능 제공 → Swagger / ReDoc자동 생성

실행 방법

• uvicorn app.main:app --reload
  • uvicorn : 서버 실행 프로그램
  • app.main : 모듈 파일 불러오기
  • app : 모듈 파일의 객체 선택
  • --relaod : 코드가 변경될 때 서버 자동 재시작 옵션

FastAPI 서버

__**init__**.py

• 현재 폴더를 패키지로 사용

app/main.py

• 서버의 메인 파일
• FastAPI app 객체 생성
• 다른 라우터들을 모듈로 불러와서 사용

from fastapi import FastAPI
# 라우터 폴더의 모듈 불러오기
from app.routers import user

# app 객체 생성
app = FastAPI()

# 기본 경로 조회
@app.get("/")
def root():
    return {"message": "FastAPI with MySQL"}

# 라우터 등록
app.include_router(user.router)

app/db.py

• MySQL 데이터베이스 연결 파일

import mysql.connector

# DB 연결 함수
def get_connection():
    return mysql.connector.connect(
        host="localhost",
        user="user",
        password="1234",
        database="codingon_db"
    )

routers/user.py

• user에 관한 정보에 접근하는 파일

from fastapi import APIRouter
from app.db import get_connection

# 라우터 객체 생성
router = APIRouter()

# users 정보 조회
@router.get("/users")
def get_users():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    result = cursor.fetchall()
    conn.close()
    return result

마치며

FastAPI에 대한 내용을 배우니까 지난 프로젝트에서 급하게 배우면서 사용했던 코드들이 조금씩 이해되는 것 같다. 이번 파트가 모두 끝나고 지난 프로젝트 코드를 다시 보면 꽤나 새로운 느낌이 들 것 같다.