37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
|
from sqlalchemy import create_engine, text
|
||
|
|
from sqlalchemy.orm import sessionmaker
|
||
|
|
from sqlalchemy.ext.declarative import declarative_base
|
||
|
|
import configparser
|
||
|
|
import logging
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
config_parser = configparser.ConfigParser()
|
||
|
|
config_parser.read("./server/server.conf")
|
||
|
|
host = config_parser.get('options', 'db_host')
|
||
|
|
port = config_parser.getint('options', 'db_port')
|
||
|
|
username = config_parser.get('options', 'db_user')
|
||
|
|
password = config_parser.get('options', 'db_password')
|
||
|
|
database = config_parser.get('options', 'db_name')
|
||
|
|
sqlname = config_parser.get('options', 'db_sqlname')
|
||
|
|
DATABASE_URL = f"{sqlname}://{username}:{password}@{host}:{port}/{database}"
|
||
|
|
engine = create_engine(DATABASE_URL)
|
||
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||
|
|
Base = declarative_base()
|
||
|
|
#获取数据库连接
|
||
|
|
def get_db():
|
||
|
|
db = SessionLocal()
|
||
|
|
try:
|
||
|
|
yield db
|
||
|
|
finally:
|
||
|
|
db.close()
|
||
|
|
logger.info("数据库关闭")
|
||
|
|
# 检查数据库连接
|
||
|
|
def check_database_connection():
|
||
|
|
try:
|
||
|
|
with SessionLocal() as db:
|
||
|
|
db.execute(text("SELECT 1"))
|
||
|
|
logger.info("数据库连接测试执行完毕")
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
logger.info(f"无法连接到数据库！错误: {e}")
|
||
|
|
return False
|