Flask + Ant Vue 实战（14）-peewee orm数据库简单实操

小儿来一壶枸杞酒泡茶 2021-02-26

645

什么是ORM

这部分的内容大家可以参考廖雪峰老师的博文：http://www.ruanyifeng.com/blog/2019/02/orm-tutorial.html

以下内容摘录廖雪峰老师的博文
ORM 就是通过实例对象的语法，完成关系型数据库的操作的技术，是"对象-关系映射"（Object/Relational Mapping）的缩写。
数据库的表（table） --> 类（class）
记录（record，行数据）--> 对象（object）
字段（field）--> 对象的属性（attribute）
ORM优点
数据模型都在一个地方定义，更容易更新和维护，也利于重用代码。
ORM 有现成的工具，很多功能都可以自动完成，比如数据消毒、预处理、事务等等。
它迫使你使用 MVC 架构，ORM 就是天然的 Model，最终使代码更清晰。
基于 ORM 的业务代码比较简单，代码量少，语义性好，容易理解。
你不必编写性能不佳的 SQL。
ORM缺点
ORM 库不是轻量级工具，需要花很多精力学习和设置。
对于复杂的查询，ORM 要么是无法表达，要么是性能不如原生的 SQL。
ORM 抽象掉了数据库层，开发者无法了解底层的数据库操作，也无法定制一些特殊的 SQL。

新建一个测试数据库

1.原生操作 python 数据操作库 records（requests作者的神作）

records的便捷在于执行原生SQL的方便和快捷。对返回的结果的集也做了相关的封装。

在没有的使用records之前的数据库的操作方式是：

#!/usr/bin/evn python
# coding=utf-8
from werkzeug.serving import run_simple


#-*- coding: utf-8 -*-
import psycopg2

def main(user,pwd,host,port,dbname):
    connection = "dbname=%s user=%s password=%s host=%s port=%s" % (dbname, user,pwd, host, port)

    db = psycopg2.connect(connection)
    cur = db.cursor()
    #创建表
    sql_stat = "CREATE TABLE test_class_id(id INT PRIMARY kEY NOT NULL, test_class TEXT, test_id CHAR(10))";
    cur.execute(sql_stat)
    #插入表
    sql_stat = "insert into test_class_id(id, test_class, test_id) values (1, 'a', '3')"
    cur.execute(sql_stat)

    sql_stat = "insert into test_class_id(id, test_class, test_id) values (2, 'a', '3')"
    cur.execute(sql_stat)
    #更改字段值
    sql_stat = "update test_class_id set test_class='b' where id=1"
    cur.execute(sql_stat)

    db.commit()

if __name__ == "__main__":
    user = 'postgres'
    pwd = '123456'
    host = 'localhost'
    port = '5432'
    dbname = 'AntFlaskDB'
    main(user, pwd, host, port, dbname)
    print("Done~~")

执行后的结果：

改用records 安装records

pip install records

链接PostgreSQL的几种方式

"""PostgreSQL"""
# default
db = records.Database('postgresql://scott:tiger@localhost/mydatabase')
# psycopg2
db = records.Database('postgresql+psycopg2://scott:tiger@localhost/mydatabase')
# pg8000
db = records.Database('postgresql+pg8000://scott:tiger@localhost/mydatabase')

创建表：

    import records

    # psycopg2username:password@localhost
    db = records.Database('postgresql+psycopg2://postgres:123456@localhost:5432/AntFlaskDB')
    rows = db.query('CREATE TABLE ceshi(id INT PRIMARY kEY NOT NULL, test_class TEXT, test_id CHAR(10))')

查询数据：

#!/usr/bin/evn python
# coding=utf-8
from werkzeug.serving import run_simple

# -*- coding: utf-8 -*-

if __name__ == "__main__":

    import records

    # psycopg2username:password@localhost
    db = records.Database('postgresql+psycopg2://postgres:123456@localhost:5432/AntFlaskDB')
    rows = db.query("select * from test_class_id")
    for row in rows:
        print(row)

db.query可以操作一切

甚至还可以直接导出相关结果集

PS:依赖pandas包，需要安装一下

pandas,
tabli
pyyaml
xlrd
xlwt 如：



if __name__ == "__main__":

    import records

    # psycopg2username:password@localhost
    db = records.Database('postgresql+psycopg2://postgres:123456@localhost:5432/AntFlaskDB')
    rows = db.query("select * from test_class_id")
    # 查看结果集
    print(rows.dataset)
    print(rows.export('csv'))
    print(rows.export('yaml'))
    print(rows.export('json'))
    # Pandas DataFrame
    rows.export('df')
    # excel
    with open('report.xls', 'wb') as f:
        f.write(rows.export('xls'))

2.ORM 操作SQLAlchemy和Peewee

这里我的就不对SQLAlchemy说明了，大家经常用估计就是这个的多。Peewee是我的当前项目使用到的。

2.1安装Peewee

pip install Peewee

2.2根据已存在的数据表生成对象模型

(AntFlask) D:\code\python\local_python\AntFlask>cd backend

(AntFlask) D:\code\python\local_python\AntFlask\backend> python -m pwiz -e postgresql -H localhost -p 5432 -u postgres -P 123456 AntFlaskDB > AntFlaskDB_model.py
Password:

(AntFlask) D:\code\python\local_python\AntFlask\backend>

执行命令输入密码：

python -m pwiz -e postgresql -H localhost -p 5432 -u postgres -P 123456 AntFlaskDB > AntFlaskDB_model.py

查看结果：AntFlaskDB_model.py详细内容为：

from peewee import *

database = PostgresqlDatabase('AntFlaskDB', **{'host': 'localhost', 'port': 5432, 'user': 'postgres', 'password': '123456'})

class UnknownField(object):
    def __init__(self, *_, **__): pass

class BaseModel(Model):
    class Meta:
        database = database

class Ceshi(BaseModel):
    test_class = TextField(null=True)
    test_id = CharField(null=True)

    class Meta:
        table_name = 'ceshi'

class TestClassId(BaseModel):
    test_class = TextField(null=True)
    test_id = CharField(null=True)

    class Meta:
        table_name = 'test_class_id'

然后我们的就可以直接使用peewee进行相关的查询数据表的操作了！如：示例

from peewee import *

database = PostgresqlDatabase('AntFlaskDB', **{'host': 'localhost', 'port': 5432, 'user': 'postgres', 'password': '123456'})


class UnknownField(object):
    def __init__(self, *_, **__): pass


class BaseModel(Model):
    class Meta:
        database = database


class Ceshi(BaseModel):
    test_class = TextField(null=True)
    test_id = CharField(null=True)

    class Meta:
        table_name = 'ceshi'


class TestClassId(BaseModel):
    test_class = TextField(null=True)
    test_id = CharField(null=True)

    class Meta:
        table_name = 'test_class_id'


if __name__ == '__main__':
    # 新增记录信息
    # TestClassId(id=3,test_class="cesjoisjd", test_id='122').save()
    #  查询所有的记录信息
    print([v for v in TestClassId.select().dicts()])
    # 根据主键ID查询
    print(TestClassId.get_by_id(1).test_class)
    # 根据主键ID查询
    print(TestClassId.get_or_none(TestClassId.test_class=='b').test_class)

修改为连接池的方式获取我们的我们的数据库的操作对象

dbhelper.py 链接数据库对象


#!/usr/bin/evn python
# coding=utf-8

from playhouse.pool import PooledPostgresqlExtDatabase

from peewee import *


db_config_const = {
    'db_name': 'AntFlaskDB',
    'db_user': 'postgres',
    'db_pass': '123456',
    'db_host': 'localhost',
    'db_port': 5432,
    'max_connections': 60,
    'stale_timeout': 300,
    'timeout': 20
}


database = PooledPostgresqlExtDatabase(
    db_config_const.get('db_name'),
    max_connections=db_config_const.get('max_connections'),
    stale_timeout=db_config_const.get('stale_timeout'),  # 5 minutes.
    timeout=db_config_const.get('timeout'),
    **{'user': db_config_const.get('db_user'), 'host': db_config_const.get('db_host'),
       'password': db_config_const.get('db_pass'), 'port': db_config_const.get('db_port')})

def GetSession():
    return database

def GetSessionAtomic():
    return database.atomic()

def session_scope_atomic():
    return GetSessionAtomic()
    # if session.is_closed():
    #     # session.connection_context()
    #     session.connect()



from contextlib import contextmanager


@contextmanager
def session_scope():
    session = GetSession()
    if session.is_closed():
        # session.connection_context()
        session.connect()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


def session_scope2():
    session = GetSession()
    with session.manual_commit():
        session.begin()  # Begin transaction explicitly.
        try:
            yield session
        except:
            session.rollback()  # Rollback -- an error occurred.
            raise
        else:
            try:
                session.commit()  # Attempt to commit changes.
            except:
                session.rollback()  # Error committing, rollback.
                raise


# cursor = GetSession.execute_sql('SELECT * FROM series(?, ?, ?)', (0, 5, 2))
# # for value, in cursor:
# #     print(value)

操作方式：

from peewee import *




from peewee import *

# database = PostgresqlDatabase('zyxadminsystem', **{'host': 'localhost', 'port': 5432, 'user': 'postgres', 'password': '123456'})

from backend.models.dbhelper import database, session_scope,session_scope_atomic
from peewee import *
from playhouse.shortcuts import model_to_dict, dict_to_model

# 自定义的
import json
class JSONField(TextField):
    def db_value(self, value):
        return json.dumps(value)

    def python_value(self, value):
        if value is not None:
            return json.loads(value)

class UnknownField(object):
    def __init__(self, *_, **__): pass

class BaseModel(Model):
    class Meta:
        database = database



class UnknownField(object):
    def __init__(self, *_, **__): pass


class BaseModel(Model):
    class Meta:
        database = database


class Ceshi(BaseModel):
    test_class = TextField(null=True)
    test_id = CharField(null=True)

    class Meta:
        table_name = 'ceshi'


class TestClassId(BaseModel):
    test_class = TextField(null=True)
    test_id = CharField(null=True)

    class Meta:
        table_name = 'test_class_id'


if __name__ == '__main__':
    # 新增记录信息
    # TestClassId(id=3,test_class="cesjoisjd", test_id='122').save()
    #  查询所有的记录信息
    with session_scope():
        print([v for v in TestClassId.select().dicts()])
        # 根据主键ID查询
        print(TestClassId.get_by_id(1).test_class)
        # 根据主键ID查询
        print(TestClassId.get_or_none(TestClassId.test_class == 'b').test_class)

END

关于PEEWEE的使用，涉及的内容比较多，这里主要是介绍怎么反向生产模型和使用with的方式进行数据库的开关处理。

小钟同学 | 文【原创】【转载请联系本人】| QQ：308711822

数据库

文章转载自小儿来一壶枸杞酒泡茶，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

Flask + Ant Vue 实战 （14）-peewee orm数据库简单实操