SQLAlchemy 操作数据库

SQLAlchemy为Python提供了不同数据库的统一接口，采用ORM的方式操作数据库，简洁优雅

一、安装

直接通过pip安装即可

pip install sqlalchemy

二、连接数据库

这里用小巧的sqlite来做测试

from sqlalchemy import create_engine# 创建连接引擎，这里的engine是lazy模式创建，直到第一次被使用才真实创建
# echo=True表示会用logger的方式打印传到数据库的SQL
engine = create_engine('sqlite:///./test.db', echo=True)

其他数据库连接方法，格式如下：

数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名

# 1.1 sqlite内存
engine = create_engine('sqlite:///:memory:', echo=True)
# 1.2 sqlite文件
engine = create_engine('sqlite:///./test.db', echo=True)# 2.1 MySQL default
engine = create_engine('mysql://user:passwd@localhost/mydatabase')
# 2.2 mysql-python
engine = create_engine('mysql+mysqldb://user:passwd@localhost/mydatabase')
# 2.3 MySQL-connector-python
engine = create_engine('mysql+mysqlconnector://user:passwd@localhost/mydatabase')
# 2.4 OurSQL
engine = create_engine('mysql+oursql://user:passwd@localhost/mydatabase')# 3.1 PostgreSQL default
engine = create_engine('postgresql://user:passwd@localhost/mydatabase')
# 3.2 psycopg2
engine = create_engine('postgresql+psycopg2://user:passwd@localhost/mydatabase')
# 3.3 pg8000
engine = create_engine('postgresql+pg8000://user:passwd@localhost/mydatabase')# 4.1 Oracle default
engine = create_engine('oracle://user:passwd@127.0.0.1:1521/mydatabase')
# 4.2 cx_oracle
engine = create_engine('oracle+cx_oracle://user:passwd@mydatabase')# 5.1 MS SQL pyodbc
engine = create_engine('mssql+pyodbc://user:passwd@mydatabase')
# 5.2 pymssql
engine = create_engine('mssql+pymssql://user:passwd@hostname:port/dbname')

可以看到，SQLAlchemy把数据库的连接和数据库的操作分离开来，向上提供统一接口

三、表结构

ORM中，表格对应一个类，所有类都要继承自Base基类

from sqlalchemy.ext.declarative import declarative_base# 基类
Base = declarative_base()# 定义User对象
class User(Base):"""Users table"""# 表的名字__tablename__ = 'users'__table_args__ = {'sqlite_autoincrement': True}# 表结构id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(32), nullable=False)age = Column(Integer, default=0)password = Column(String(64), unique=True)

四、操作

操作需要通过session来进行，增删改结束后，需要commit，查询不用

# 删除
session = DBSession()
duser = session.query(User).filter(User.id==2).delete()
session.commit()
session.close()#查询
session = DBSession()
quser = session.query(User).filter(User.id==4).one()
print('name:',quser.name)
session.close()

操作体验和很多ORM一致

五、完整测试代码

参考http://www.itfanr.cc/2017/01/06/use-sqlalchemy-by-python/

# coding=utf-8from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker# 创建连接引擎
# echo=True表示会用logger的方式打印传到数据库的SQL
engine = create_engine('sqlite:///./test.db', echo=True)# 表格对象基类
Base = declarative_base()# 创建会话的类
DBSession = sessionmaker(bind=engine)# 表格类
class User(Base):"""User table"""__tablename__ = 'users'  # 表名__table_args__ = {'sqlite_autoincrement': True}# 表结构id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(32), nullable=False)  # 有些数据库允许不指定String的长度age = Column(Integer, default=0)password = Column(String(64), unique=True)class Blog(Base):"""docstring for Blog"""__tablename__ = 'blogs'id = Column(Integer, primary_key=True)title = Column(String(100))desc = Column(String(500))class Tips(Base):"""docstring for tips"""__tablename__ = 'tips'id = Column(Integer, primary_key=True)name = Column(String(32))# 新增一条记录数据
def new_user():# 创建会话对象session = DBSession()new_user = User(name='Jery',password='123')session.add(new_user)  # 只是添加在session中，没有写到数据库session.commit()session.close()# 新增多条数据
def add_more_user():session = DBSession()session.add_all([User(name='guanyu', age=4, password='11111'),User(name='zhangfei', password='2233'),User(name='zhenji', password='44556')])session.commit()session.close()# 新增数据含中文,只要用unicode的字符即可
def add_user_for_zh():session = DBSession()new_user = User(name=u'关羽', password='12322233')session.add(new_user)session.commit()session.close()# 查询
def query_user():session = DBSession()q_user = session.query(User).filter(User.id == 4).one()  # one表示结果有且仅有一个print('name', q_user.name)session.close()  # 查询不用commit，如果不commit，会自动执行rollback#
def delete_user():session = DBSession()deleted_num = session.query(User).filter(User.id > 3).delete()print(deleted_num)session.commit()session.close()# c测试
def test_user():session = DBSession()# merge方法，如果存在就修改，不存在就插入（只判断主键，不判断unique列）t1 = session.query(User).filter(User.name == 'Jery').first()t1.age = 34session.merge(t1)session.commit()# 获取第2-3项users = session.query(User)[1:3]for u in users:print(u.name)session.close()# 执行sql语句
def sql_user():s = DBSession()# 不能用 `?` 的方式来传递参数 要用 `:param` 的形式来指定参数# s.execute('INSERT INTO users (name, age, password) VALUES (?, ?, ?)',('bigpang',2,'1122121'))# 这样执行报错# s.execute('INSERT INTO users (name, age, password) VALUES (:aa, :bb, :cc)',({'aa':'bigpang2','bb':22,'cc':'998'}))# s.commit()# 这样执行成功res = s.execute('select * from users where age=:aaa', {'aaa': 4})# print(res['name'])  # 错误# print(res.name)    # 错误# print(type(res))   # 错误for r in res:print(r['name'])s.close()# 执行sql语句
def sql_user2():# **传统 connection方式**# 创建一个connection对象，使用方法与调用python自带的sqlite使用方式类似# 使用with 来创建 conn，不需要显示执行关闭连接# with engine.connect() as conn:#  res=conn.execute('select * from users')#  data=res.fetchone()#  print('user is %s' %data[1])# 与python自带的sqlite不同，这里不需要 cursor 光标，执行sql语句不需要commit。如果是增删改，则直接生效，也不需要commit.# **传统 connection 事务**with engine.connect() as conn:trans = conn.begin()try:r1 = conn.execute("select * from users")print(r1.fetchone())r2 = conn.execute("insert into users (name,age,password) values (?,?,?)", ('tang', 5, '133444'))print(r2)trans.commit()except:trans.rollback()raise# **session**session = DBSession()session.execute('select * from users')session.execute('insert into users (name,age,password) values (:name,:age,:password)',{"name": 'dayuzhishui', 'age': 6, 'password': '887'})# 注意参数使用dict，并在sql语句中使用:key占位# 如果是增删改，需要 commitsession.commit()# 用完记得关闭，也可以用 withsession.close()if __name__ == "__main__":# 删除全部数据库Base.metadata.drop_all(engine)# 创建表格，如果已经存在，则不创建Base.metadata.create_all(engine)# 删除指定的表格Blog.__table__.drop(engine)# 新增数据new_user()# 新增多条数据add_more_user()# 新增数据含中文add_user_for_zh()# 查询query_user()# 删除delete_user()# 测试test_user()# 直接执行SQLsql_user()sql_user2()

用pycharm运行后的结构