repository/MySQL/Config_Database_MySQL.py

from sqlalchemy import create_engine, URL, text
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from sqlalchemy.exc import OperationalError
from dotenv import load_dotenv
import os
load_dotenv()
import os, sys
import os
from dotenv import load_dotenv

# Load biến môi trường từ file .env

current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(current_dir, '..', '..')) 
from dotenv import load_dotenv, find_dotenv

# Tự động tìm file .env gần nhất trong cây thư mục cha
load_dotenv(find_dotenv(), override=True)

MYSQL_USER_NAME = os.getenv('DB_USER')
MYSQL_PASSWORD = os.getenv('DB_PASSWORD')
MYSQL_PORT = os.getenv('DB_PORT')
MYSQL_DATABASE = os.getenv('DB_NAME')
MYSQL_HOST = os.getenv('DB_HOST')
MYSQL_PORT = int(MYSQL_PORT) if MYSQL_PORT else 3306 

import os
from urllib.parse import quote

password = os.getenv("DB_PASSWORD") 
DB_PASSWORD = quote(password)

Base = DeclarativeBase()

# def get_db_engine():
#     """ Tạo kết nối đến MySQL """
#     dsn = URL.create(
#         drivername="mysql+pymysql",
#         username=MYSQL_USER_NAME,
#         # password=MYSQL_PASSWORD,
#         host=MYSQL_HOST,
#         port=MYSQL_PORT,
#         database=MYSQL_DATABASE
#     )
#     return create_engine(dsn)

# Connect server MYSQL
def get_db_engine():
    dsn = URL.create(
        drivername="mysql+pymysql",
        username=MYSQL_USER_NAME,
        password=MYSQL_PASSWORD,
        host=MYSQL_HOST,
        port=MYSQL_PORT,
        database=MYSQL_DATABASE
    )
    connect_args = {
            "ssl_verify_cert": True,
            "ssl_verify_identity": True,
            "ssl_ca": 'ca.pem',
        }
    return create_engine(
        dsn,
        connect_args=connect_args,
    )


# echo=True để debug SQL logs

# # Khởi tạo engine
# engine = get_db_engine()

# # Tạo session factory
# SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# def test_connection():
#     """ Kiểm tra kết nối MySQL """
#     try:
#         with engine.connect() as connection:
#             result = connection.execute(text("SELECT 1"))  # Dùng text() để tránh lỗi ObjectNotExecutableError
#             print("✅ Kết nối MySQL thành công!", result.scalar())  # result.scalar() lấy giá trị đầu tiên
#     except OperationalError as e:
#         print("❌ Kết nối MySQL thất bại!", str(e))

# # Chạy test kết nối
# test_connection()