Tornado 操作数据库

191次阅读

共计 5586 个字符，预计需要花费 14 分钟才能阅读完成。

与 Django 或者 Flask 相比 Tornado 没有自带的 ORM 对于数据库需要去适配我们使用 MySQL 数据库

说明

在 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 和 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()