python实现sqlalchemy的使用概述

from sqlalchemy import create_engine  # 引擎
from sqlalchemy.orm import sessionmaker  # 创建orm的会话池，orm和sql均可以管理对象关系型数据库，需要绑定引擎才可以使用会话，
# 创建连接
engine = create_engine("mysql+pymysql://root:1234;@127.0.0.1/test", # 需要安装mysql和pymysql的模块，用户名:密码@ip地址/某个数据库
  #echo=True,# 打印操作对应的SQL语句
  pool_size=8, # 连接个数
  pool_recycle=60*30 # 不使用时断开
  )
# 创建session
DbSession = sessionmaker(bind=engine)  # 会话工厂，与引擎绑定。
session = DbSession()# 实例化
session.close()  # 关闭会话

# 导入定义类需要的模块
from sqlalchemy.ext.declarative import declarative_base # 调用sqlalchemy的基类
from sqlalchemy import Column, Index, distinct, update  # 指定字段属性，索引、唯一、DML
from sqlalchemy.types import * # 所有字段类型

# 创建库表类型
Base = declarative_base()  # 调用sqlalchemy的基类
class Users(Base):
 '''继承基类'''
 __tablename__ = "users"# 数据表的名字
 __table_args__ = {'extend_existing': True}  # 当数据库中已经有该表时，或内存中已声明该表，可以用此语句重新覆盖声明。
 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 # 声明需要调用的特征，可以只声明数据库中表格列的子集
  
Base.metadata.create_all(engine) # 表生效：将所有定义的类，使用引擎创建，此时可以在数据库中看到这些表。

# 创建库表类型
Base = declarative_base()  # 调用sqlalchemy的基类
class model_data(BASE):  
 '''创建数据库表类：模型所需的基本字段'''
 __abstract__ = True # 辅助sqlAlchemy实现类的继承，自动继承属性，省去super()
 __table_args__ = {'extend_existing': True}  # 若表的声明在内存中已存在，则重新声明表的名称，不然会报错
 ai_xdr_id = Column(BigInteger(), primary_key=True, unique=True, autoincrement= True)
 ai_sdk_id = Column(BigInteger())
class TrainData(model_data): # 训练集表
 '''继承model_data的属性，并将表的名字定义为：'xxx_train_data'存入数据库 '''
 __tablename__ = 'xxx_train_data'
 
class DevData(model_data):# 开发集表
 '''表的名字定义为：'xxx_dev_data' '''
 __tablename__ = 'xxx_dev_data'
class TestData(model_data):  # 测试集表
 __tablename__ = 'xxx_test_data'
Base.metadata.create_all(engine) # 表生效：将所有定义的类，使用引擎创建，此时可以在数据库中看到这些表。

# 增加数据
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()
# 当上述语句出现执行错误时，需要执行回滚语句，才能继续操作
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)# 若未在类中声明，则无法访问数据库中该表的属性。

class DataAccessLayer:# 数据连接层、定义了连接和关闭。
 '''数据连接层、定义了连接和关闭。'''
 def __init__(self):
  self.ENGINE = None  # 引擎
  self.SESSION = None # 会话
  self.conn_string = "mysql+pymysql://root:1234;@127.0.0.1/test"  ## 需要安装mysql和pymysql的模块，用户名:密码@ip地址/某个数据库
 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):
  '''断开时，关闭引擎。'''
  self.ENGINE.close()
def df_save_db(df,tablename):
 '''将数据集DataFrame保存到数据库'''
 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')