阿里云-云小站(无限量代金券发放中)
【腾讯云】云服务器、云数据库、COS、CDN、短信等热卖云产品特惠抢购

Tornado 操作数据库

283次阅读
没有评论

共计 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()

      先查询结果,再将查询的结果删除,如果查询的结果为空,则删除会报错

正文完
星哥玩云-微信公众号
post-qrcode
 0
星锅
版权声明:本站原创文章,由 星锅 于2022-05-26发表,共计5586字。
转载说明:除特殊说明外本站文章皆由CC-4.0协议发布,转载请注明出处。
【腾讯云】推广者专属福利,新客户无门槛领取总价值高达2860元代金券,每种代金券限量500张,先到先得。
阿里云-最新活动爆款每日限量供应
评论(没有评论)
验证码
【腾讯云】云服务器、云数据库、COS、CDN、短信等云产品特惠热卖中