Fastapi 와 DB 연결

Fastapi 와 DB 연결 (mysql)

1. 라이브러리 및 환경 설정

import mysql.connector
from mysql.connector import Error
from contextlib import contextmanager
from dotenv import load_dotenv
import os

mysql.connector: MySQL 데이터베이스에 연결하고 쿼리를 실행하기 위한 라이브러리.
Error: 데이터베이스 연결 및 작업 중 발생하는 오류를 처리하기 위한 예외 클래스.
contextmanager: Python의 context manager를 생성하기 위한 데코레이터.
load_dotenv: .env 파일에서 환경 변수를 로드하는 함수.
os: 환경 변수 접근을 위한 표준 라이브러리.

2. 환경 변수 로드

load_dotenv()

.env 파일을 로드하여 데이터베이스 연결 정보를 가져옴.

3. 데이터베이스 연결 정보

DB_HOST = os.getenv("DB_HOST")
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_NAME = os.getenv("DB_NAME")
DB_PORT = os.getenv("DB_PORT")

환경 변수에서 데이터베이스 연결 정보(호스트, 유저, 비밀번호, DB 이름, 포트)를 가져옴.

4. 데이터베이스 연결 함수

def get_db_connection():
    try:
        connection = mysql.connector.connect(
            host=DB_HOST,
            user=DB_USER,
            password=DB_PASSWORD,
            database=DB_NAME,
            port=DB_PORT
        )
        if connection.is_connected():
            return connection
    except Error as e:
        print(f"Error connecting to MySQL platform: {e}")
        return None

MySQL에 연결하고 성공 시 connection 객체를 반환.
연결 실패 시 오류를 출력하고 None을 반환.

5. 데이터베이스 커서 및 연결 관리 함수

@contextmanager
def get_db_cursor():
    connection = get_db_connection()
    if connection:
        cursor = connection.cursor(dictionary=True)
        try:
            yield cursor, connection
        finally:
            cursor.close()
            connection.close()
    else:
        raise Exception("Database connection failed")

contextmanager 데코레이터를 사용하여 커서와 연결을 관리.
데이터베이스 연결과 커서를 열고, 작업 후 자동으로 닫음ㅁ.
실패 시 예외를 발생.
dictionary=True: 쿼리 결과를 딕셔너리 형식으로 반환하도록 설정.

6. 연결 테스트 함수

def test_db_connection():
    try:
        connection = get_db_connection()
        
        if connection:
            cursor = connection.cursor()
            cursor.execute("SELECT 1")
            result = cursor.fetchone()
            if result:
                print("Database connection successful!")
            else:
                print("Database query failed.")
            cursor.close()
            connection.close()
        else:
            print("Failed to connect to the database.")
    except Exception as e:
        print(f"Error: {e}")

데이터베이스에 연결한 후, 간단한 쿼리(SELECT 1)를 실행.
결과가 반환되면 연결 성공 메시지를 출력.

7. 실행 코드

test_db_connection()

위에서 정의한 test_db_connection 함수를 실행하여 데이터베이스 연결을 테스트.

주요 기능 요약

.env 파일에서 DB 연결 정보를 읽음.
DB 연결을 시도하고 get_db_connection 함수로 연결 객체를 반환.
contextmanager를 사용해 커서와 연결을 안전하게 관리.
test_db_connection으로 연결 및 간단한 쿼리 테스트.

전체코드

import mysql.connector
from mysql.connector import Error
from contextlib import contextmanager
from dotenv import load_dotenv
import os

# .env 파일 로드
load_dotenv()

# 환경 변수에서 데이터베이스 연결 정보 가져오기
DB_HOST = os.getenv("DB_HOST")
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_NAME = os.getenv("DB_NAME")
DB_PORT = os.getenv("DB_PORT")

# 데이터베이스 연결 설정
def get_db_connection():
    try:
        connection = mysql.connector.connect(
            host=DB_HOST,
            user=DB_USER,
            password=DB_PASSWORD,
            database=DB_NAME,
            port=DB_PORT
        )
        if connection.is_connected():
            return connection
    except Error as e:
        print(f"Error connecting to MySQL platform: {e}")
        return None

# 커서와 연결을 함께 반환
@contextmanager
def get_db_cursor():
    connection = get_db_connection()
    if connection:
        cursor = connection.cursor(dictionary=True)
        try:
            yield cursor, connection
        finally:
            cursor.close()
            connection.close()
    else:
        raise Exception("Database connection failed")



# 연결 확인용 간단한 코드
def test_db_connection():
    try:
        # 연결 시도
        connection = get_db_connection()
        
        if connection:
            # 연결이 성공하면 간단한 쿼리 실행
            cursor = connection.cursor()
            cursor.execute("SELECT 1")
            result = cursor.fetchone()  # 결과 확인
            if result:
                print("Database connection successful!")
            else:
                print("Database query failed.")
            cursor.close()
            connection.close()  # 연결 종료
        else:
            print("Failed to connect to the database.")
    except Exception as e:
        print(f"Error: {e}")

# 테스트 실행
test_db_connection()