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

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

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

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

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

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

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

class Users(Base):
    '''繼承基類(lèi)'''
    __tablename__ = "users"                     # 數(shù)據(jù)表的名字
    __table_args__ = {'extend_existing': True}  # 當(dāng)數(shù)據(jù)庫(kù)中已經(jīng)有該表時(shí)，或內(nèi)存中已聲明該表，可以用此語(yǔ)句重新覆蓋聲明。
    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ù)庫(kù)中表格列的子集
        
Base.metadata.create_all(engine)                # 表生效：將所有定義的類(lèi)，使用引擎創(chuàng)建，此時(shí)可以在數(shù)據(jù)庫(kù)中看到這些表。

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

class model_data(BASE):  
    '''創(chuàng)建數(shù)據(jù)庫(kù)表類(lèi)：模型所需的基本字段'''
    __abstract__ = True                         # 輔助sqlAlchemy實(shí)現(xiàn)類(lèi)的繼承，自動(dòng)繼承屬性，省去super()
    __table_args__ = {'extend_existing': True}  # 若表的聲明在內(nèi)存中已存在，則重新聲明表的名稱，不然會(huì)報(bào)錯(cuò)
    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ù)庫(kù) '''
    __tablename__ = 'xxx_train_data'
    
class DevData(model_data):   # 開(kāi)發(fā)集表
    '''表的名字定義為：'xxx_dev_data' '''
    __tablename__ = 'xxx_dev_data'

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

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

# 增加數(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)上述語(yǔ)句出現(xiàn)執(zhí)行錯(cuò)誤時(shí)，需要執(zhí)行回滾語(yǔ)句，才能繼續(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)   # 若未在類(lèi)中聲明，則無(wú)法訪問(wèn)數(shù)據(jù)庫(kù)中該表的屬性。

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

    def connect(self):
        '''連接時(shí)建立引擎和會(huì)話。'''
        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):
        '''斷開(kāi)時(shí)，關(guān)閉引擎。'''
        self.ENGINE.close()

def df_save_db(df,tablename):
    '''將數(shù)據(jù)集DataFrame保存到數(shù)據(jù)庫(kù)'''
    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')