python實現(xiàn)sqlalchemy的使用概述

from sqlalchemy import create_engine                          # 引擎
from sqlalchemy.orm import sessionmaker                       # 創(chuàng)建orm的會話池，orm和sql均可以管理對象關(guān)系型數(shù)據(jù)庫，需要綁定引擎才可以使用會話，

# 創(chuàng)建連接
engine = create_engine("mysql+pymysql://root:1234;@127.0.0.1/test", # 需要安裝mysql和pymysql的模塊，用戶名:密碼@ip地址/某個數(shù)據(jù)庫
                       #echo=True,         # 打印操作對應(yīng)的SQL語句
                       pool_size=8,       # 連接個數(shù)
                       pool_recycle=60*30 # 不使用時斷開
                       )

# 創(chuàng)建session
DbSession = sessionmaker(bind=engine)  # 會話工廠，與引擎綁定。
session = DbSession()                  # 實例化

session.close()  # 關(guān)閉會話

# 導(dǎo)入定義類需要的模塊
from sqlalchemy.ext.declarative import declarative_base       # 調(diào)用sqlalchemy的基類
from sqlalchemy import Column, Index, distinct, update        # 指定字段屬性，索引、唯一、DML
from sqlalchemy.types import *                                # 所有字段類型

# 創(chuàng)建庫表類型
Base = declarative_base()  # 調(diào)用sqlalchemy的基類

class Users(Base):
    '''繼承基類'''
    __tablename__ = "users"                     # 數(shù)據(jù)表的名字
    __table_args__ = {'extend_existing': True}  # 當(dāng)數(shù)據(jù)庫中已經(jīng)有該表時，或內(nèi)存中已聲明該表，可以用此語句重新覆蓋聲明。
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True)
    #email = Column(String(64))

    def __init__(self, name, email):
        self.name = name
        self.email = email                      # 聲明需要調(diào)用的特征，可以只聲明數(shù)據(jù)庫中表格列的子集
        
Base.metadata.create_all(engine)                # 表生效：將所有定義的類，使用引擎創(chuàng)建，此時可以在數(shù)據(jù)庫中看到這些表。

# 創(chuàng)建庫表類型
Base = declarative_base()  # 調(diào)用sqlalchemy的基類

class model_data(BASE):  
    '''創(chuàng)建數(shù)據(jù)庫表類：模型所需的基本字段'''
    __abstract__ = True                         # 輔助sqlAlchemy實現(xiàn)類的繼承，自動繼承屬性，省去super()
    __table_args__ = {'extend_existing': True}  # 若表的聲明在內(nèi)存中已存在，則重新聲明表的名稱，不然會報錯
    ai_xdr_id = Column(BigInteger(), primary_key=True, unique=True, autoincrement= True)
    ai_sdk_id = Column(BigInteger())

class TrainData(model_data): # 訓(xùn)練集表
    '''繼承model_data的屬性，并將表的名字定義為：'xxx_train_data'存入數(shù)據(jù)庫 '''
    __tablename__ = 'xxx_train_data'
    
class DevData(model_data):   # 開發(fā)集表
    '''表的名字定義為：'xxx_dev_data' '''
    __tablename__ = 'xxx_dev_data'

class TestData(model_data):  # 測試集表
    __tablename__ = 'xxx_test_data'

Base.metadata.create_all(engine)                # 表生效：將所有定義的類，使用引擎創(chuàng)建，此時可以在數(shù)據(jù)庫中看到這些表。

# 增加數(shù)據(jù)
add_user = Users("test3", "test123@qq.com")
session.add(add_user)
session.commit()

# add_users = Users(("test", "test123@qq.com"),('a','b')))
# session.add(add_users)
# session.commit()

# 當(dāng)上述語句出現(xiàn)執(zhí)行錯誤時，需要執(zhí)行回滾語句，才能繼續(xù)操作
session.rollback()

delete_users = session.query(Users).filter(Users.name == "test").first()
if delete_users:
    session.delete(delete_users)
    session.commit()
    
session.query(Users).filter(Users.name == "test").delete()
session.commit()

# 改
session.query(Users).filter_by(id=1).update({'name': "Jack"})

users = session.query(Users).filter_by(name="Jack").first()
users.name = "test"

users = session.query(Users).filter_by(id=5).all()
for item in users:
    print(item.name)
    print(item.email)   # 若未在類中聲明，則無法訪問數(shù)據(jù)庫中該表的屬性。

class DataAccessLayer:# 數(shù)據(jù)連接層、定義了連接和關(guān)閉。
    '''數(shù)據(jù)連接層、定義了連接和關(guān)閉。'''
    def __init__(self):
        self.ENGINE = None                 # 引擎
        self.SESSION = None                # 會話
        self.conn_string = "mysql+pymysql://root:1234;@127.0.0.1/test"  ## 需要安裝mysql和pymysql的模塊，用戶名:密碼@ip地址/某個數(shù)據(jù)庫

    def connect(self):
        '''連接時建立引擎和會話。'''
        self.ENGINE = create_engine(self.conn_string, encoding='utf-8',isolation_level="AUTOCOMMIT", connect_args={'connect_timeout': 7200})
        # self.ENGINE = create_engine(self.conn_string, encoding='utf-8',connect_args={'connect_timeout': 7200})
        self.SESSION = sessionmaker(bind=self.ENGINE)()

    def disconnect(self):
        '''斷開時，關(guān)閉引擎。'''
        self.ENGINE.close()

def df_save_db(df,tablename):
    '''將數(shù)據(jù)集DataFrame保存到數(shù)據(jù)庫'''
    db_ac = DataAccessLayer()
    db_ac.connect()
    conn = db_ac.ENGINE.connect()
    df.to_sql(name=tablename, con=conn, if_exists='append', index=False)
    conn.close()
    print('%s updated.'%tablename)

df = pd.read_csv('traindata_jiangsu_donghai.csv')
df_save_db(df,'traindata_jiangsu_donghai')