共计 5586 个字符,预计需要花费 14 分钟才能阅读完成。
一、概述
与 Django 或者 Flask 相比 Tornado 没有自带的 ORM 对于数据库需要去适配 我们使用 MySQL 数据库
二、torndb
-
说明
在 Tornado3.0 版本一起 提供 tornado.database 模块用来操作 MySQL 数据库 而从 3.0 版本开始 此模块就被独立出来 作为 torndb 包单独提供 torndb 只是对 MySQLdb 的简单封装 不支持 python3(torndb 是支持 python2.x.x 版本 不支持 Python3 先正常安装 然后在进行修改
-
更改步骤:
-
先正常安装
pip install torndb
-
pip show torndb 查看安装的路径
-
把 lucky 老师的 torndb 替换到我们 torndb 安装的路径的位置
-
-
连接初始化
我们需要在应用启动时创建一个数据库连接实例 供各个 RequestHandler 进行使用 我们可以在构造 Application 的时候 创建一个数据库实例 并作为其属性 而 RequestHandler 可以通过 self.application 获取其属性 进而操作数据库实例
代码如下:
import tornado.web import tornado.ioloop import torndb class IndexHandler(tornado.web.RequestHandler): def get(self): self.write('操作 MySQL 数据库') class Application(tornado.web.Application): def __init__(self,handlers,**settings): super(Application, self).__init__(handlers,**settings) # 创建一个全局 MySQL 链接实例 供 handler 使用 self.db = torndb.Connection(host='127.0.0.1', # 主机 database='torndb', # 数据库名称 user='lucky', # 用户名 password='123456', # 密码 ) if __name__ == '__main__': app = Application([(r'/',IndexHandler), ], debug=True, autoreload=True ) app.listen(8000) tornado.ioloop.IOLoop.current().start()
-
创建操作使用的表
语句如下
mysql> create table if not exists user( -> id int unsigned primary key auto_increment, -> username varchar(10) default 'lucky', -> age tinyint default 18, -> info varchar(100) default '个人简介' -> );
-
执行语句
- execute(query) 返回最后一条自增字段值
- execute_rowcount(query) 返回影响的行数
示例
execute 的使用:
class IndexHandler(tornado.web.RequestHandler): def get(self): # 执行语句 res = self.application.db.execute("insert into user values(null,'lucky',18,' 我是帅气的 lucky 老师 ')") print(res) self.write('操作 MySQL 数据库')
execute_rowcount 的使用:
def get(self): # 执行语句 res = self.application.db.execute_rowcount("select * from user") print(res) self.write('操作 MySQL 数据库')
-
查询语句
- get(query) 返回单行结果或 None,如果出现多行则报错 返回值是一个字典的对象 支持对象属性的获取和关键字索引
- query(query) 返回多行结果 torndb.Row 的列表
示例
get 的使用:
class IndexHandler(tornado.web.RequestHandler): def get(self): # 查询语句 res = self.application.db.get("select * from user where id=1") print(res['info']) print(res.username) self.write('操作 MySQL 数据库')
query 的使用:
class IndexHandler(tornado.web.RequestHandler): def get(self): # 查询语句 res = self.application.db.query("select * from user") print(res) self.write('操作 MySQL 数据库')
三、sqlalchemy
-
安装 sqlalchemy 和 pymysql
- pip install sqlalchemy
- pip install pymysql
-
连接数据库
from sqlalchemy import create_engine config = {'HOST': '', 'USERNAME': '', 'PASSWORD': '', 'PORT': '', 'DATABASE': '', 'PARAMS': '' } DB_URL = 'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}?{PARAMS}' engine = create_engine(DB_URL.format(**config)) if __name__ == '__main__': conc = engine.connect() result = conc.execute('select 1') print(result.fetchone())
如果连接成功,则控制台会出现(1,)的结果
-
数据库建模
-
建立 base 类
from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base config = {'HOST': '', 'USERNAME': '', 'PASSWORD': '', 'PORT': '', 'DATABASE': '', 'PARAMS': '' } # 连接数据连接 URL DB_URL = 'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}?{PARAMS}' # 连接数据库 engine = create_engine(DB_URL.format(**DB_CONFIG)) # 模型类基类 Base = declarative_base(engine)
-
使用 base 类并建立表
from settings import Base from sqlalchemy import Column, Integer, String, DateTime, Boolean from datetime import datetime class User(Base): __tablename__ = 'user' # 指定表名 id = Column(Integer, autoincrement=True, primary_key=True) username = Column(String(20)) password = Column(String(20)) create_time = Column(DateTime, default=datetime.now()) is_login = Column(Boolean, default=False, nullable=False) if __name__ == '__main__': Base.metadata.create_all() # 创建表
创建的表名必须为该类的__tablename__属性,Colum 新建一个字段,然后给 Colum 传参来进行约束
-
-
Column 的常用参数:
- default: 默认值,可以传一个函数体,default 的值等于这个函数体执行后返回的值
- nullable:是否可为空
- primary_key:是否为主键
- unique:是否唯一
- autoincrement:是否自增长
- onupdate:更新的时候执行的函数,和 default 一样,可以传一个函数体
- name:该属性在数据库中的字段的映射,默认是属性名
-
常用的数据类型
-
Integer:整形
-
Float:浮点类型
-
Boolean:布尔
-
DECIMAL:定点类
DECIMAL 第一个参数为整数位的个数,第二位参数为小数位的个数
-
Enum:枚举类型
Enum 可以借助 python3 自带的 enum 包来实现更加简便
-
Date:传递 datetime.date()
-
DateTime:传递 datetime.datetime()
-
Time:传递 datetime.time() 进去
-
String:字符串型, 使用时需要制定长度
-
Text:文本类型
-
LONGTEXT:长文本类型
-
-
简单的增删改查
-
增
from sqlalchemy import create_engine from settings import DB_CONFIG from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 连接数据连接 URL DB_URL = 'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}?{PARAMS}' # 连接数据库 engine = create_engine(DB_URL.format(**DB_CONFIG)) # 模型基类 Base = declarative_base(engine) # 增删改查 创建会话 Session = sessionmaker(engine) # 实例 session = Session()
与上面不同的是, 新建了一个 Session 对象,然后将这个对象实例化,接下来的数据库操作都用到这个实例对象来操作
表结构
from config import Base from sqlalchemy import Column, Integer, String, DateTime, Boolean from datetime import datetime class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(20)) password = Column(String(20)) create_time = Column(DateTime, default=datetime.now()) is_login = Column(Boolean, default=False, nullable=False)
-
添加
添加一条
from models import User from config import session def add(): user = User(username='lucky', password='123456') session.add(user) session.commit() if __name__ == '__main__': add()
添加多条
from models import User from config import session def add(): session.add_all([User(username='lucky1', password='123456'), User(username='lucky2', password='123456'), User(username='lucky3', password='123456'), ]) session.commit() if __name__ == '__main__': add()
-
查
from models import User from config import session def search(): session.query(User).all() # 查找所有 result = session.query(User).first().username # 查找第一个 result = session.query(User).filter(User.username=='lucky').first().password print(result) if __name__ == '__main__': search()
-
改
更改的数据以字典的键值对的形式传入
from models import User from config import session def update(): session.query(User).filter(User.username == 'ivy').update( {User.username: 'lucky_boy', User.password: '123654', } ) session.commit() if __name__ == '__main__': update()
-
删
from models import User from config import session def delete(): result = session.query(User).filter(User.username == 'lucky').first() session.delete(result) session.commit() if __name__ == '__main__': delete()
先查询结果,再将查询的结果删除,如果查询的结果为空,则删除会报错
-