SQLAlchemy 的配置鏈接文件
以下為SQLAlchemy的配置文件，請勿直接復(fù)制，建議手敲，畢竟網(wǎng)頁復(fù)制會使得代碼變形。
Python可是非常注意代碼的格式的！
from sqlalchemy import Column,create_engine,MetaData,Table,func,event,exc
from sqlalchemy.orm import sessionmaker,scoped_session
from sqlalchemy.types import *
from sqlalchemy.ext.declarative import declarative_base
from urllib.parse import quote_plus
import os
# 數(shù)據(jù)庫配置
dbConfig = {
? ? "dbname":"數(shù)據(jù)庫名",
? ? "dbusername":"數(shù)據(jù)庫用戶名",
? ? "dbpasswd":"數(shù)據(jù)庫密碼",
? ? "dbhost":"數(shù)據(jù)庫地址",
? ? "dbport":3306,
? ? "dbcharset":"數(shù)據(jù)庫編碼格式"
}
base = declarative_base()
engine = create_engine(
? ? # 形成原則
? ? # mysql+pymysql://數(shù)據(jù)庫用戶名:數(shù)據(jù)庫密碼@服務(wù)器地址:端口號/數(shù)據(jù)庫名?charset=字符集
? ? # 如果密碼中帶了@ 這個特殊字符，需要將quote_plus 這個方法包裹字符串 或者
? ? "mysql+pymysql://fastapiblog:"+quote_plus('Abcd@123456')+"@localhost:3306/fastapi?charset=utf8mb4",
? ? # 數(shù)據(jù)庫連接池，默認(rèn)為5，建議調(diào)整是調(diào)整到CPU內(nèi)核數(shù)x2,因為調(diào)整到太大
? ? pool_size=10,
? ? # 這個入?yún)⑹菢?biāo)識是否打印SQL建議在開發(fā)環(huán)境開啟。
? ? echo=True,
? ? # MySQL 在默認(rèn)情況下鏈接8小時沒有任何操作，會自動斷開鏈接，此后會出現(xiàn)MySQL has gone away的錯誤。
? ? # 因此需要設(shè)置這個參數(shù)來回收鏈接。這個鏈接的單位是s也就是設(shè)置3000s之后，將會被sqlalchemy給回收。
? ? pool_recycle=3000,
? ? # 連接池ping，當(dāng)從連接池拿鏈接的時候，會向數(shù)據(jù)庫發(fā)送一個select 1（類似）的操作以確定是否數(shù)據(jù)庫是否存活。
? ? pool_pre_ping=True
)
metadata = MetaData(engine)
import datetime
@event.listens_for(engine, "connect")
def connect(dbapi_connection, connection_record):
? ? connection_record.info['pid'] = os.getpid()
@event.listens_for(engine, "checkout")
def checkout(dbapi_connection, connection_record, connection_proxy):
? ? pid = os.getpid()
? ? if connection_record.info['pid'] != pid:
? ? ? ? connection_record.connection = connection_proxy.connection = None
? ? ? ? raise exc.DisconnectionError(
? ? ? ? ? ? "Connection record belongs to pid %s, "
? ? ? ? ? ? "attempting to check out in pid %s" %
? ? ? ? ? ? (connection_record.info['pid'], pid)
? ? ? ? )
'''
請將表建在下方
'''
# 固定形式需要繼承base 否則sqlalchemy 是無法執(zhí)行查詢等操作的
class UserTable(base):
? ? # 固定形式，必須在這里定義表名，否則無法查詢到表
? ? __tablename__ = "user"
? ? # 定義int結(jié)構(gòu)的主鍵，自增
? ? id = Column(Integer,primary_key=True,autoincrement="auto")
? ? create_time = Column(DATETIME)
? ? update_time = Column(DATETIME)
? ? username = Column(String(64))
? ? password = Column(String(64))
? ? mobile = Column(String(20))
? ? email = Column(String(64))
? ? status = Column(String(16))
'''
請將表建在上方
'''
SessionType = scoped_session(sessionmaker(bind=engine, expire_on_commit=False))
def GetSession():
? ? return SessionType()
from contextlib import contextmanager
@contextmanager
def session_scope():
? ? session = GetSession()
? ? try:
? ? ? ? yield session
? ? ? ? session.commit()
? ? except:
? ? ? ? session.rollback()
? ? ? ? raise
? ? finally:
? ? ? ? session.close()
# 檢查數(shù)據(jù)庫是否存在該表。
def check_table_exist(tablename):
? ? with session_scope() as sess:
? ? ? ? r = sess.execute(
? ? ? ? ? ? "select TABLE_NAME from INFORMATION_SCHEMA.TABLES where TABLE_NAME='" + tablename + "' AND TABLE_SCHEMA='"+dbConfig["dbname"]+"';")
? ? ? ? if len(r.fetchall()) >= 1:
? ? ? ? ? ? return True
? ? ? ? else:
? ? ? ? ? ? return False
if __name__ == "__main__":
? ? print(quote_plus('Abcd@123456'))
? ? # 這個方法能夠創(chuàng)建所有繼承了base的表。
? ? base.metadata.create_all(engine)
