Python ORM | peewee

2022 年 08 月 06 日

137 次浏览

5243字数

python使用过程中，难免要用到一些关系型数据库，例如sqlite, mysql, postgresql，而增删查改过程中，字符串的拼接着实让人头大

如果你有用过Django web后端框架，自带的ORM组件简化了数据库的操作，在常规数据库的操作方式上，又抽象了一层，把一个个的数据库文件，一张张的表，抽象为一个个的对象，一个个的 python类；增删查改等等操作，抽象为一个个的函数和属性，以面向对象编程的思路和方式来操作数据库。

这里推荐一个独立的Python ORM库----peewee，不需要Django环境。

github项目地址:

https://github.com/coleifer/peewee

文档地址：

http://docs.peewee-orm.com/en/latest/peewee/quickstart.html

如果你有使用Django ORM的经验，peewee非常容易理解和上手

创建数据库文件和表：

from peewee import *
import datetime


db = SqliteDatabase('ip_monitor_db.db')

class ip_monitor_db(Model):
    manage_ip = CharField(unique=True)
    online_or_not = BooleanField(default=True)
    monitor_or_not = BooleanField(default=True)
    sent_or_not = BooleanField(default=False)
    last_sent_time = DateTimeField(default=datetime.datetime.now)
    

    class Meta:
        database = db # if there are many databases ,This model uses the "testing.db" database.

class Person(Model):
    name = CharField()
    birthday = DateField()

    class Meta:
        database = db # This model uses the "ip_monitor_db.db" database.

if __name__ == '__main__':
    ############################# create DB and Tables
    db.connect()
    db.create_tables([ip_monitor_db,Person])

插入

     ############################### insert items （2 ways）
     ip1 = ip_monitor_db(manage_ip = 'baidu.com')
     ip1.save()
     ip2 = ip_monitor_db(manage_ip = '114.114.114.114', monitor_or_not = False)
     ip2.save()
     ip3 = ip_monitor_db.create(manage_ip = '100.64.0.1')

查询

    ############################# query items
    # 查询所有
     all = ip_monitor_db.select()
     print(all)
    # 条件查询（2 ways）
     all1 = ip_monitor_db.get(ip_monitor_db.manage_ip == 'baidu.com')
     print(all1)
     all2 = ip_monitor_db.select().where(ip_monitor_db.online_or_not == True)
     all2 = ip_monitor_db.select().where(ip_monitor_db.online_or_not == True).get() # only get the first one element when there are many targets
   # 如果有多条查询结果，可以迭代该结果
    for ip in ip_monitor_db.select():
        print(ip.manage_ip)

    #
    query = ip_monitor_db.select().where(ip_monitor_db.online_or_not == True)
    for element in query:
        print(element.manage_ip, element.last_sent_time)

更新

    ####################################update items
     all4 = ip_monitor_db.get(ip_monitor_db.manage_ip == 'baidu.com')
     all4.manage_ip = '114.114.114.114'
     all4.monitor_or_not = False
     all4.save()
     print(all4.manage_ip)
     print(all4.monitor_or_not)

删除

     ###########################delete items
     all3 = ip_monitor_db.delete().where(ip_monitor_db.monitor_or_not == False).execute()
     print(all3)

其他还有更多，联合查找，排序，外键等等参考官方文档

附上常规的字符串拼接操作数据库方式，sqlite3为例

from asyncio.windows_events import NULL
import sqlite3
from datetime import datetime, timedelta


def create_tab():
    conn = sqlite3.connect('ip_monitor_db.sqlite3')
    c = conn.cursor()
    #  创建表
    sql = '''
        CREATE TABLE ip_monitor_db(
        ID INTEGER PRIMARY KEY  AUTOINCREMENT,
		manage_ip text NOT NULL UNIQUE,
        online_or_not bool NOT NULL,
		monitor_or_not bool NOT NULL,
        sent_or_not bool NOT NULL,
        last_sent_time datetime NOT NULL)
        '''
    c.execute(sql)
    conn.commit()
    c.close()
    conn.close()


def insert_tab(manage_ip, online_or_not = True, monitor_or_not= True, last_sent_time = datetime.now(), sent_or_not = False):
    conn = sqlite3.connect('ip_monitor_db.sqlite3')
    c = conn.cursor()
    #  插入表
    sql = '''INSERT INTO ip_monitor_db(manage_ip, online_or_not, monitor_or_not, last_sent_time, sent_or_not) VALUES (?,?,?,?,?)'''
    c.execute(sql, (manage_ip, online_or_not, monitor_or_not, last_sent_time, sent_or_not))
    conn.commit()
    c.close()
    conn.close()

def update_tab(manage_ip, online_or_not=NULL, last_sent_time = NULL, sent_or_not = NULL):
    conn = sqlite3.connect('ip_monitor_db.sqlite3')
    c = conn.cursor()
    #  更新表
    sql = '''UPDATE ip_monitor_db SET online_or_not = ?, last_sent_time = ?, sent_or_not = ? WHERE manage_ip = ?'''
    c.execute(sql, (online_or_not, last_sent_time, sent_or_not, manage_ip))
    conn.commit()
    c.close()
    conn.close()

def select_tab(column='*'):
    conn = sqlite3.connect('ip_monitor_db.sqlite3')
    c = conn.cursor()
    #  查询表
    sql = f'''SELECT {column} FROM ip_monitor_db'''
    results = c.execute(sql)
    results_all = results.fetchall()
    return results_all
    conn.commit()
    c.close()
    conn.close()



if __name__ == '__main__':
    #create_tab()
    #insert_tab(manage_ip='baidu.com')
    #insert_tab(manage_ip='114.114.114.114')
    #insert_tab(manage_ip='100.64.0.1')
    print(select_tab('manage_ip'))
    print(select_tab('manage_ip'))

以上

Python ORM | peewee