python操作三大主流数据库(3)python操作mysql③python操作mysql的orm工具sqlaichemy安装配置和使用

python操作mysql③python操作mysql的orm工具sqlaichemy安装配置和使用

手册地址： http://docs.sqlalchemy.org/en/rel_1_1/orm/index.html 
安装
D:\software\source_tar>pip install SQLALchemy

检测是否安装成功
D:\software\source_tar>python
Python 3.5.2 (v3.5.2:4def2a2901a5, Jun 25 2016, 22:18:55) [MSC v.1900 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import sqlalchemy

mysql的orm库SQLALchemy的操作

#coding:utf-8

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

'''
id int primary key auto_increment,
title varchar(200) not null,
content varchar(2000) not null,
tpes varchar(10) not null,
image varchar(300) null,
author varchar(20) null,
view_count int default 0,
created_at datetime null,
is_valid smallint default 1
'''
# 创建对象的基类
Base = declarative_base()
# 初始化数据库连接，注意要接上charset=utf8否则中文无法支持
engine = create_engine("mysql://root:@localhost/news?charset=utf8")
# 创建DBSession类型
DBSession = sessionmaker(bind=engine)

# 定义News对象
class News(Base):
__tablename__ = 'news'
id = Column(Integer, primary_key=True)
title = Column(String(200), nullable=False)
content = Column(String(2000), nullable=False)
types = Column(String(10), nullable=False)
image = Column(String(300),)
author = Column(String(20),)
view_count = Column(Integer)
created_at = Column(DateTime)
is_valid = Column(Boolean)

'''
# 简单测试
# 如果表不存在就创建表
Base.metadata.create_all(engine)

# 创建session对象:
session = DBSession()
# 创建新User对象，新增一条测试数据
news01 = News(title='标题1', content = 'content01', types = 'baidu',
image = '/static/img/01.jpg', author = 'jack', view_count = 3)
# 添加到session:
session.add(news01)
# 提交即保存到数据库:
session.commit()
# 关闭session:
session.close()
'''

# orm的测试类
class OrmTest(object):
# 初始化创建session
def __init__(self):
self.session = DBSession()

# 添加数据
def add_one(self):
new_obj = News(
title = '标题20180202',
content = '内容20180202',
types = '百家'
)
self.session.add(new_obj)
self.session.commit()
return new_obj

# 添加多条数据
def add_more(self):
add_list = []
for i in range(10):
new_obj = News(title='标题%s'%str(i),content='内容%s'%str(i),types='百家%s'%str(i))
self.session.add(new_obj)
add_list.append(new_obj)
self.session.commit()
return add_list

# 删除数据
def delete_data(self):
data = self.session.query(News).get(51)
self.session.delete(data)
self.session.commit()

# 修改单条数据
def update_one(self, _id):
obj = self.session.query(News).get(_id)
if obj:
obj.is_valid = 0
self.session.add(obj)
self.session.commit()
return True

return False

# 修改多条数据
def update_data(self):
# filter_by的使用方法
# data_list = self.session.query(News).filter_by(is_valid = 0)

# filter的使用方法
data_list = self.session.query(News).filter(News.id > 45)
for data in data_list:
print(data.title)
data.is_valid = 1
self.session.add(data)
self.session.commit()

# 获取一条数据
def get_one(self):
return self.session.query(News).get(1)

# 获取多条数据
def get_more(self):
return self.session.query(News).filter_by(is_valid = 1)

def main():
obj = OrmTest()
# rst = obj.add_one()
# print('id:%s, title:%s,content:%s,types = %s' % (rst.id,rst.title,rst.content,rst.types))

# 添加多条数据
# rst = obj.add_more()
# for _new in rst:
#     print('id:{0},title{1},content:{2}'.format(_new.id,_new.title,_new.content))

# 测试删除
# obj.delete_data()

# 修改单条数据
# print(obj.update_one(50))

# 修改多条数据
obj.update_data()

# 测试获取一条数据的函数
# rst = obj.get_one()
# print(rst.title)

# 获取多条数据
# rst = obj.get_more()
# for _news in rst:
#     print('news id: %s, title:%s, content:%s' % (_news.id,_news.title,_news.content))

if __name__ == "__main__":
main()