CSTOS

人生苦短 Python水深

0%

Python-关系数据库

发表于 更新于 分类于
本文字数： 9.9k 阅读时长 ≈ 36 分钟

对关系数据库进行相关操作的 Python 库。如：

  • pyodbc：封装了 ODBC API，通过它可访问各种数据库
  • PyMysql：首选 Mysql 的 Python 驱动
    • MySQLdb：备用，明明是 MysqlClient，导入的是 MySQLdb，因为是改自它吧
  • pymssql：操作 sql server
  • psycopg：PostgreSQL
  • cx_oracle：操作 Oracle
  • SQLAlchemy：既支持原生 SQL，又支持 ORM 的工具
  • peewee：轻量 python orm 库
  • dataset：提供了一个简单的抽象层， 无需完整的 ORM 可省大量 SQL 语句，可像 JSON 和 NoSQL 一样操作数据库
  • Gadfly：提供了一个完整的 SQL 环境
  • sandman2：号称不用一行代码，用 API 操作数据库，不咋地，OUT

概述

Python 能够通过遵循 PythonDB-API（数据库应用程序编程接口）规范的模块与 Microsoft SQLServer，Oracle，Sybase，DB2，MySQL、SQLite 等数据库进行通信。

  • 数据库选择，首选 MySQL，其次是其自带的 sqlite。

基本常识

  • SQL：通用数据语言，不要低估 sql 的力量（相比之下，pandas 还是有所限制的）
  • 表、行、字段…..等关系数据库基本知识，详见 数据库篇。
  • Python 开发相关：
    • Connection： 要操作关系数据库，首先需要连接到数据库，一个数据库连接称为 Connection；
    • **Cursor：**要执行 SQL 语句，需要游标，
      Python 定义了一套操作数据库的 API 接口，任何数据库要连接到 Python，只需要提供符合 Python 标准的数据库驱动即可。
  • Python 的 DB-API:
    • Connection 和 Cursor 对象，打开后记得关闭
    • 使用 Cursor 对象执行 insert，update，delete 语句时，执行结果由 rowcount 返回影响的行数，就可以拿到执行结果。
    • 使用 Cursor 对象执行 select 语句时，通过 featchall() 可以拿到结果集。结果集是一个 list，每个元素都是一个 tuple，对应一行记录。
      如果 SQL 语句带有参数，那么需要把参数按照位置传递给 execute() 方法，有几个 ? 占位符就必须对应几个参数，例如：
1
cursor.execute('select * from user where name=? and pwd=?', ('abc', 'password'))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
rpm -q MySQL-python
import MySQLdb
conn = MySQLdb.connect(user='root',passwd='',host='127.0.0.1')
cur = conn.cursor() #创建游标
conn.select_db('student')
cur.execute('insert into student(name,age,gender) value('chen',18,'m')')
sqli= 'insert into student(name,age,gender) value({},{},'{})'.format(name,age,gender)
cur.execute(sqli,('chen2',18,'m'))
cur.executemany(sqli,[('chen3',18,'m'),('chen4',18,'m'),('chen5',18,'m')]) #插入多条值
cur.execute('delete from student where id =20')
cur.execuet('update studen set name='zhang' where id='12')
cur.execute("select * from student")  #只显示数据量
cur.fetchone()
cur.scroll(1,'absolute')
cur.fetchmany(cur.execute("select * from student"))
cur.close()
conn.close()

# cursor 游标
execute()：执行单条语句
executemany()：执行多条语句
fetchone()：获取第一行
fetchmany(size)：获取下几(size)行
fetchall()：获取剩下所有的行
scroll(num,mode)：移动游标
rowcount：计算 corsor 的行数
description：返回字段信息
close()：关闭游标

连接

1
2
3
4
5
6
7
8
conn.begin() # 开启事务
conn.commit() # 提交事务
conn.cursor() # 创建游标
conn.ping()  # 检查链接是否存活，会重新发起链接
conn.rollback() # 加滚事务
conn.close()  # 关闭连寻章摘句
conn.select_db() #　选择数据 库
conn.show_warnings() # 查看 warnings 信息

游标

1
2
3
4
# 创建连接
# 创建连接
# cursor=conn.cursor() # 一般的游标
# # 游标设置为字典类型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# sql="select id,name,content,start_time,end_time,importance,emergency,completion,insert_time,update_time,remarks from t_01001_pending"

    num1=cur.execute(sql)  # 执行并返回影响行数

aa=cur.fetchone() #第一行，游标到第一行后
    bb=cur.fetchmany(3) #第二到四行，游标到第四行后
    cc=cur.fetchall()

    # 游标移动

    cur.scroll(0, mode='absolute') 绝对位置移动，以 0 作为起点，移动游标，第一次参数只能是大于或等于 0
    cur.scroll(-1, mode='relative')相对位置移动 ，以游标目前的位置作为起点，然后根据第一参数进行移动。负数向左移，正数向右移

    # conn.commit() # 提交新建或修改的数据

    import sqlite3
    con=sqlite3.connect('population.db')
    cur=con.cursor()
    cur.execut()

cur.execute('''CREATE TABLE stuent(name TEXT NOT NULL,age INTEGER,
                    class TEXT,CONSTRAINT CountryKey PEIMIRY KEY (name))''')
cur.execute('INSERT INTO student VALUES ("chen",18,"计算机科学与技术")')
    con.commint()
    cur.execute('CREATE TABLE PopByCountry(Region TEXT, Country TEXT, Population INEGER)')
    countries = [("Eastern Asia", "China", 1285238), ("Eastern Asia", "DPR Korea", 24056), ("Eastern Asia", "Hong Kong (China)", 8764), ("Eastern Asia", "Mongolia", 3407), ("Eastern Asia", "Republic of Korea", 41491), ("Eastern Asia", "Taiwan", 1433), ("North America", "Bahamas", 368), ("North America", "Canada", 40876), ("North America", "Greenland", 43), ("North America", "Mexico", 126875), ("North America", "United States", 493038)]
    for c in countries:
        cur.execute('INSERT INTO PopByCountry VALUES (?, ?, ?)', (c[0], c[1],
                    c[2]))
    con.commit()
    cur.execute('DELETE FROM student WHERE name="chen"')
    cur.execute('SELECT name FROM student WHERE age >18')
    cur.fetchall()
    cur.execute('SELECT age,class FROM student WHERE name={}'.format(name))
    age = cur.fetchone()[0]
    cur.commit()

ORM

ORM：Object-Relational Mapping：对象 - 关系映射系统，将关系数据库的表结构映射到对象上。

  • SQLAlchemy
  • 优点：
    • 提高开发效率，降低开发成本
    • 使开发更加对象化：将数据库表的一行记录与一个对象互相做自动转换
    • 可移植
    • 可方便地引入数据缓存之类的附加功能
  • 缺点
    • 自动化进行关系数据库的映射需要消耗系统性能（但基本可忽略不计）
    • 处理多表联查、where 条件复杂时，ORM 的语法会变得复杂。

sqlite3

sqlite3：【标准库】Python 内置的 SQLite3，可直接使用

  • 提供了一个可以使用稍微非标准的 SQL 语法更新和访问的持久数据库。
  • 特点：
    • C 写的，轻量，体积小，可嵌入，数据库为一个文件
    • 无服务器、零配置、事务性的 SQL 数据库引擎
    • 不能承受高并发
    • 常用于桌面和移动应用

sqlite3：Python 自带的轻量级数据库模块

  • SQLite：用 C 写的小型、快速、功能齐全的关系灵气库引擎
  • 设计目标：嵌入式
    • 最大支持 128TiB 数据

数据类型

NULL：空值：None

INTEGER：带符号的整数：int/long

REAL：浮点数：float

TEXT：文本字符串：str/unicode

BLOB：blob 数据：buffer

布尔值被存储为整数 0（false）和 1（true）

  • 索引不应该使用在较小的表上。
  • 索引不应该使用在有频繁的大批量的更新或插入操作的表上。
  • 索引不应该使用在含有大量的 NULL 值的列上。
  • 索引不应该使用在频繁操作的列上。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import sqlite3
conn = sqlite3.connect('test.db') # 创建连接，若无 test.db，将创建
cursor = conn.cursor()
cursor.execute('create table user(id varchar(20) primary key,name varchar(20))')
cursor.execute('insert into user(id,name) values("1","Chen")')
cursor.rowcount
cursor.close() # 关闭 cursor
conn.close() # 关闭 connection


# 查
conn=sqlite3.connect('test.db')
cursor=conn.cursor()
cursor.execut('select name form user where id = "1"')
values=cursor.fetchall()
print(values)
cursor.close()
conn.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47


def max_sql_variables():
    """Get the maximum number of arguments allowed in a query by the current
    sqlite3 implementation. Based on `this question
    `_

    Returns
    -------
    int
            inferred SQLITE_MAX_VARIABLE_NUMBER
    """
    import sqlite3

    db = sqlite3.connect(':memory:')
    cur = db.cursor()
    cur.execute('CREATE TABLE t (test)')
    low, high = 0, 100000
    while (high - 1) > low:
        guess = (high + low) // 2
        query = 'INSERT INTO t VALUES ' + ','.join(['(?)' for _ in range(guess)])
        args = [str(i) for i in range(guess)]
        try:
            cur.execute(query, args)
        except sqlite3.OperationalError as e:
            if "too many SQL variables" in str(e):
                high = guess
            else:
                raise
        else:
            low = guess
    cur.close()
    db.close()
    return low


SQLITE_MAX_VARIABLE_NUMBER = max_sql_variables()
print(SQLITE_MAX_VARIABLE_NUMBER)


# sqlite3插入有数量限制

with config.DB.atomic() as txn:
    size = (SQLITE_MAX_VARIABLE_NUMBER // len(line_list[0])) - 1
    # remove one to avoid issue if peewee adds some variable
    for i in range(0, len(line_list), size):
        config.T_Anno.insert_many(line_list[i : i + size]).execute()

pyodbc

pyodbc：封装了 ODBC API，通过它可访问各种数据库

pymysql

PyMysql：首选 Mysql 的 Python 驱动

MySQLdb：备用，明明是 MysqlClient，导入的是 MySQLdb，因为是改自它吧

  • Web 世界中使用最广泛的数据库服务器。
  • Mysql 相比 SQLite 高并发性能强，占用内存多。
  • MySQL 的工具：首选 pymysql，其次是 mysql-connector-python，但 mysqllient 说是有极致性能，mysql-connector-python 是 MySQL 官方提供的，让人纠结…
  • MySQL 的 SQL 占位符是 %s。
  • mysql-connector-python 安装的时候需要给 pip 命令加上参数 --allow-external
    • pip install --allow-external mysql-connector-python

基本使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import pymysql
conn=pymysql.connect(host='127.0.0.1', # 主机名
                  port=3306, # 端口号
                     user='chen',  # 登录用户名
                    passwd='123456',  # 登录密码
                     database='', # 连接数据库
                    charset='utf8mb4'， # 连接字符串
                     db='',  # 数据库别名
                    connect_timeout=10# 建立连接超时时间，默认为 10,最小为 1,最大 31536000
                    )
cursor = conn.cursor()  # 创建游标
sql_create ='create table user (id varchar(20) primary key, name varchar(20))'
sql_insert = 'insert into 员工资料 values(6,"董事长","十一郎",123)'  # 新增 SQL 语句
sql_update = "update 员工资料 set 密码 = 111 WHERE ID = 6"  # 修改 SQL 语句
sql_delete = "delete from 员工资料 where ID = 6"  # 删除 SQL 语句
try:
    cur.execute(sql_insert)  # 执行新增 SQL 语句
    cur.execute(sql_update)  # 执行修改 SQL 语句
    cur.execute(sql_delete)  # 执行删除 SQL 语句
    conn.commit()  # 提交事务
except Exception as e:
    conn.rollback()  # 如果发生错误，则回滚事务
finally:
    cursor.close()  # 关闭游标 cursor
    conn.close()  # 关闭数据库连接

连接

1
2
3
4
5
6
7
8
9
10
11
12
13
import pymysql
# 简洁
conn=pymysql.connect('localhost',3306,'chen','123456','test',utf8mb4)
# 优雅
db_config={
    'host':'127.0.0.1',
    'port':3306,
    'user':'chen',
    'password':'123456',
    'db':'test',
    'charset':'utf8mb4',
}
conn =pymysql.connect(**db_config)

插入

1
2
3
4
5
6
7
try:
    with conn.cursor() as cur:
        sql='insert into test(name,age) values({},{})'
        cur.execute(sql.format)
    conn.commit()
finally:
    conn.close()

查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
try:
    with conn.cursor() as cur:
        sql="select name,age from student"
        cur.execute(sql)
        result=cur.fetchone()
        print(result)
    conn.commit()
finally:
    conn.close(0)
# 打开游标
try:
    cur = db.cursor()
    sql = "select * from {table_name}"
    cur.execute(sql.format(table_name="student")) #执行
    results=cur.fetchall()  #获取查到的数据
    #最近一次 execute 返回数据的行数或影响的行数：
 row=you.rowcount()
    for i in results:
        id=i[0]
        name=i[1]
        gender=i[2]
        print(id,name,gender)
except Exception as e:
    raise e
finally:
    db.close()

    cursor.rowcount
cursor.execute('select * from user where id = %s', ('1',))
cursor.executemany(sql,[(...),{...}])
values = cursor.fetchall()
values
[('1', 'Michael')]

pymssql

pymssql：操作 sql server

psyscopg

psycopg：PostgreSQL

cx_oracle

cx_oracle：操作 Oracle

SQLAlchemy

SQLAlchemy：既支持原生 SQL，又支持 ORM 的工具

  • 最知名的 ORM 框架之一
  • 定义 class，将表的内容用 Python 数据结构表示，list 下多个 tuple 表示多行数据。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from sqlalchemy import Column,String,create_engine
from sqlalchemy.orm improt sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base= declarative_base() # 创建对象基类
class User(Base):
    __tablename__='user'
    id=Column(String(20),primary_key=True)
    name=Column(String(20))

# 初始化数据库连接：'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
engine=create_engine('mysql+mysqlconnector://root:password@localhost:3306/test')
DBSession=sessionmaker(bind=engine)

session = DBSession() # 创建 session 对象,即当前的数据库的连接
new_user=User(id='5',name='Chen') # 创建新 User 对象
session.add(new_user) # 添加
session.commit() # 提交到数据库
session.close()  # 关闭 session

# 创建 Query 查询，filter 是 where 条件，最后调用 one()返回唯一行，如果调用 all()则返回所有行:
user = session.query(User).filter(User.id=='5').one()
# 打印类型和对象的 name 属性:
print('type:', type(user))
print('name:', user.name)
# 关闭 Session:
session.close()
  • ORM：将数据库表的行与相应的对象建立关联，互相转换。
1
2
3
4
5
6
7
8
9
10
11
12
class User(Base):
    __tablename__ = 'user'
    id = Column(String(20), primary_key=True)
    name = Column(String(20))
    # 一对多:
    books = relationship('Book')
class Book(Base):
    __tablename__ = 'book'
    id = Column(String(20), primary_key=True)
    name = Column(String(20))
    # “多”的一方的 book 表是通过外键关联到 user 表的:
    user_id = Column(String(20), ForeignKey('user.id'))
  • 主要组件：
    • SQLAlchemy ORM：构建在 Core 之上，以提供可选的 对象关系映身功能。Python 类映射到数据库表、会话的对象持久化机制。
      • Object Relation Mapper（ORM）
    • SQLAlchemy Core：数据库工具包的基础架构。包括：数据库连接、查询、结果交互及程序化构建 SQL 语句的工具。
      • Schema/Types
      • SQL Expression Language
      • Engine
        • Connection Pooling
        • Dialect
    • DBAPI
  • 使用版本 2.0 以上 sqlalchemy.__version__

建立连接 - 引擎

  • engine 对象：充当数据库连接的中心来源，提供工厂、连接池
    • 延迟连接：engine 在首次 create_engine() 返回时并未连接到数据，它在首次要求对数据库执行任务时连接。
1
2
3
4
5
6
from sqlalchemy import create_engine
engine=create_engine("sqlite+pysqlite:///:memory:",echo=True)
# sqlite: 数据库类型
# pysqlite: DBAPI: 与特定数据库交互的第三方驱动程序
# /:memory：使用仅在内存中的数据库
#　echo:将其发出的所有 SQL 记录到将写入标准输出的 Python 记录器中

事务

  • engine 的目的是通过提供 Connection 对象来连接数据库
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# text: 将SQL语句编写成文本SQL
from sqlalchemy import text

with engine.connect() as conne:
    result=conn.execute(text("select 'hello world'"))
    print(result.all())
    
    
    conn.execute(text("CREATE TABLE some_table (x int, y int)"))
    conn.execute(
        text("INSERT INTO some_table (x,y) VALUES (:x,:y)"),
        [{"x":1,"y":1},{"x":2,"y":4}],
    )
    conn.commit() # 此时才会提交
# 连接释放时，会输出 ROLLBACK 以结束事务

with engine.begin() as conn:
    pass

查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
with engine.connect() as conn:
    # 结果行的可迭代对象，像Python命名元组一样工作
    result=conn.execute(text("SELECT x,y FROM some_table"))
    # 属性名称
    for row in result:
        print(f'x: {row.x},y:{row.y}')
        
    # 直接元组赋值
    for x,y in result:
        pass
       
    # 使用索引 
    for row in result:
        x=row[0]
        
    # 映射访问
    for dict_row in result.mappings(): # 将Result转换为 MappingResult对象
        x = dict_row["x"]
        y = dict_row["y"]
  • 参数传入
1
2
3
4
5
6
7
8
9
10
11
12
with engine.connect() as conn:
    result = conn.execute(text("SELECT x, y FROM some_table WHERE y > :y"), {"y": 2})
    for row in result:
        print(f"x: {row.x}  y: {row.y}")
 
# 发送多个参数       
with engine.connect() as conn:
    conn.execute(
        text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
        [{"x": 11, "y": 12}, {"x": 13, "y": 14}],
    )
    conn.commit()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from sqlalchemy.orm import Session

stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y")
with Session(engine) as session:
    result = session.execute(stmt, {"y": 6})
    for row in result:
        print(f"x: {row.x}  y: {row.y}")
        
with Session(engine) as session:
    result = session.execute(
        text("UPDATE some_table SET y=:y WHERE x=:x"),
        [{"x": 9, "y": 11}, {"x": 13, "y": 15}],
    )
    session.commit()

DBAPI

数据库元数据

  • MetaData
  • Table
  • Column
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from sqlalchemy import MetaData
metadata_obj = MetaData()

from sqlalchemy import Table, Column, Integer, String
user_table = Table(
    "user_account",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("name", String(30)),
    Column("fullname", String),
)

# Column 对象相对于父 Table 的集合通常通过位于 Table.c 的关联数组访问
user_table.c.keys()
user_table.c.fullname


from sqlalchemy import ForeignKey
address_table = Table(
    "address",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("user_id", ForeignKey("user_account.id"), nullable=False),
    Column("email_address", String, nullable=False),
)

metadata_obj.create_all(engine)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from sqlalchemy.orm import DeclarativeBase

# 声明基类引用为我们自动创建 MetData
class Base(DeclarativeBase):
    pass
    
Base.metadata
Base.registry # 中心“映射器配置”单元


from typing import List
from typing import Optional
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class User(Base): # ORM映射类，可用于ORM持久性和查询操作。
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(back_populates="user")
    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id = mapped_column(ForeignKey("user_account.id"))
    user: Mapped[User] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

Base.metadata.create_all(engine)
  • 表反射：通过读取数据的当前状态来生成Table和相关对象的过程
1
some_table = Table("some_table", metadata_obj, autoload_with=engine)

insert

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from sqlalchemy import insert
stmt = insert(user_table).values(name="spongebob", fullname="Spongebob Squarepants")
print(stmt)
# INSERT INTO user_account (name, fullname) VALUES (:name, :fullname)
with engine.connect() as conn:
    result = conn.execute(stmt)
    conn.commit()
    
result.inserted_primary_key
(1,) # 元组，因为主键可能包含多个列，这被称为复合主键


with engine.connect() as conn:
    result = conn.execute(
        insert(user_table),
        [
            {"name": "sandy", "fullname": "Sandy Cheeks"},
            {"name": "patrick", "fullname": "Patrick Star"},
        ],
    )
    conn.commit()
    
    
# 技巧
from sqlalchemy import select, bindparam
scalar_subq = (
    select(user_table.c.id)
    .where(user_table.c.name == bindparam("username"))
    .scalar_subquery()
)

with engine.connect() as conn:
    result = conn.execute(
        insert(address_table).values(user_id=scalar_subq),
        [
            {
                "username": "spongebob",
                "email_address": "spongebob@sqlalchemy.org",
            },
            {"username": "sandy", "email_address": "sandy@sqlalchemy.org"},
            {"username": "sandy", "email_address": "sandy@squirrelpower.org"},
        ],
    )
    conn.commit()
  • insert语句可以带returning
1
2
3
4
5
6
select_stmt = select(user_table.c.id, user_table.c.name + "@aol.com")
insert_stmt = insert(address_table).from_select(
    ["user_id", "email_address"], select_stmt
)
print(insert_stmt)
print(insert_stmt.returning(address_table.c.id, address_table.c.email_address))

select

  • 生成一个Select构造
  • 生成式方法，每个方法都在对象上构造更多状态
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
from sqlalchemy import select
stmt = select(user_table).where(user_table.c.name == "spongebob")
print(stmt)


with engine.connect() as conn:
    #运行结果返回行，可迭代对象
    for row in conn.execute(stmt):
        print(row)
print(select(user_table))    
print(select(user_table.c.name, user_table.c.fullname))
print(select(user_table.c["name", "fullname"]))

# ORM形式
stmt = select(User).where(User.name == "spongebob")
with Session(engine) as session:
    for row in session.execute(stmt):
        print(row)
        
print(select(User))
row = session.execute(select(User)).first()
# 方法将返回一个 ScalarResult 对象，该对象立即传递每行的第一个“列”，在本例中为 User 类的实例
user = session.scalars(select(User)).first() 
print(select(User.name, User.fullname))
# SELECT user_account.name, user_account.fullname FROM user_account

row = session.execute(select(User.name, User.fullname)).first()
session.execute(
    select(User.name, Address).where(User.id == Address.user_id).order_by(Address.id)
).all()

# label: 在结果集中具有特定的名称
from sqlalchemy import func, cast
stmt = select(
    ("Username: " + user_table.c.name).label("username"),
).order_by(user_table.c.name)
with engine.connect() as conn:
    for row in conn.execute(stmt):
        print(f"{row.username}")
        
        
from sqlalchemy import text
stmt = select(text("'some phrase'"), user_table.c.name).order_by(user_table.c.name)
with engine.connect() as conn:
    print(conn.execute(stmt).all())
    
# literal_column:从文本片段中获取更多的功能,显示地表示单个列,可以标记并在子查询和其他表达式中引用.
# 在这两种情况下，当使用 text() 或 literal_column() 时，我们正在编写语法 SQL 表达式，而不是文字值。因此，我们必须包含我们想要看到的呈现的 SQL 所需的任何引号或语法。
from sqlalchemy import literal_column
stmt = select(literal_column("'some phrase'").label("p"), user_table.c.name).order_by(
    user_table.c.name
)
with engine.connect() as conn:
    for row in conn.execute(stmt):
        print(f"{row.p}, {row.name}")

where

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
print(user_table.c.name == "squidward")

print(address_table.c.user_id > 10)

print(select(user_table).where(user_table.c.name == "squidward"))

# 多次调用
print(
    select(address_table.c.email_address)
    .where(user_table.c.name == "squidward")
    .where(address_table.c.user_id == user_table.c.id)
)

from sqlalchemy import and_, or_
print(
    select(Address.email_address).where(
        and_(
            or_(User.name == "squidward", User.name == "sandy"),
            Address.user_id == User.id,
        )
    )
)

对于针对单个实体的简单“相等”比较，还有一种流行的方法称为 Select.filter_by()，它接受与列键或 ORM 属性名称匹配的关键字参数。它将根据最左侧的 FROM 子句或最后加入的实体进行过滤

1
print(select(User).filter_by(name="spongebob", fullname="Spongebob Squarepants"))

from-join

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
print(select(user_table.c.name))

print(select(user_table.c.name, address_table.c.email_address))
# 显式地指示join的左侧和右侧
print(
    select(user_table.c.name, address_table.c.email_address).join_from(
        user_table, address_table
    )
)
# 仅指示右侧
print(select(user_table.c.name, address_table.c.email_address).join(address_table))

# select_from:显式添加
print(select(address_table.c.email_address).select_from(user_table).join(address_table))

# 列子句没有足够的信息来提供 FROM 子句时,也需要使用select_from
from sqlalchemy import func
print(select(func.count("*")).select_from(user_table))


# on 
print(
    select(address_table.c.email_address)
    .select_from(user_table)
    .join(address_table, user_table.c.id == address_table.c.user_id)
)

# outer 和 full join
# LEFT OUTER JOIN
print(select(user_table).join(address_table, isouter=True))
#  FULL OUTER JOIN
print(select(user_table).join(address_table, full=True))

# SQL 也有 “RIGHT OUTER JOIN”。SQLAlchemy 不会直接呈现它；而是反转表的顺序并使用 “LEFT OUTER JOIN”。

order by-group by-having

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# 排序
print(select(user_table).order_by(user_table.c.name))
print(select(User).order_by(User.fullname.desc()))
# 分组,调用聚合函数
from sqlalchemy import func
count_fn = func.count(user_table.c.id)
print(count_fn)

# 选择用户名字段以及地址计数，适用于拥有多个地址的用户
with engine.connect() as conn:
    result = conn.execute(
        select(User.name, func.count(Address.id).label("count"))
        .join(Address)
        .group_by(User.name)
        .having(func.count(Address.id) > 1) # 应用于组内使用的聚合函数
    )
    print(result.all())

# 特别是在某些数据库后端上，是能够按已在 columns 子句中声明的表达式进行 ORDER BY 或 GROUP BY，而无需在 ORDER BY 或 GROUP BY 子句中重新声明该表达式，而是使用来自 COLUMNS 子句的列名或标签名。
from sqlalchemy import func, desc
stmt = (
    select(Address.user_id, func.count(Address.id).label("num_addresses"))
    .group_by("user_id")
    .order_by("user_id", desc("num_addresses"))
)
print(stmt)

# 别名:为表或子查询提供备用名称，可以从该名称在语句中引用它
user_alias_1 = user_table.alias()
user_alias_2 = user_table.alias()
print(
    select(user_alias_1.c.name, user_alias_2.c.name).join_from(
        user_alias_1, user_alias_2, user_alias_1.c.id > user_alias_2.c.id
    )
)

# ORM 实体别名
from sqlalchemy.orm import aliased
address_alias_1 = aliased(Address)
address_alias_2 = aliased(Address)
print(
    select(User)
    .join_from(User, address_alias_1)
    .where(address_alias_1.email_address == "patrick@aol.com")
    .join_from(User, address_alias_2)
    .where(address_alias_2.email_address == "patrick@gmail.com")
)

子查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# SQLAlchemy 使用 Subquery 对象来表示子查询，使用 CTE 来表示 CTE，通常分别从 Select.subquery() 和 Select.cte() 方法获得。 这两个对象都可以用作较大 select() 构造中的 FROM 元素。
subq = (
    select(func.count(address_table.c.id).label("count"), address_table.c.user_id)
    .group_by(address_table.c.user_id)
    .subquery()
)

print(select(subq.c.user_id, subq.c.count))

stmt = select(user_table.c.name, user_table.c.fullname, subq.c.count).join_from(
    user_table, subq
)

print(stmt)


# CTE:公用表表达式:将结果对象用作 FROM 元素，但呈现的 SQL 是非常不同的公用表表达式语法
subq = (
    select(func.count(address_table.c.id).label("count"), address_table.c.user_id)
    .group_by(address_table.c.user_id)
    .cte()
)

stmt = select(user_table.c.name, user_table.c.fullname, subq.c.count).join_from(
    user_table, subq
)

print(stmt)

CTE 构造还具有以“递归”样式使用的能力，并且在更复杂的情况下，可以从 INSERT、UPDATE 或 DELETE 语句的 RETURNING 子句中组成。 CTE 的文档字符串包含有关这些其他模式的详细信息。

在这两种情况下，子查询和 CTE 都在 SQL 级别使用“匿名”名称命名。 在 Python 代码中，我们根本不需要提供这些名称。 Subquery 或 CTE 实例的对象标识充当呈现时对象的语法标识。 可以通过将其作为 Select.subquery() 或 Select.cte() 方法的第一个参数传递来提供将在 SQL 中呈现的名称。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
subq = select(Address).where(~Address.email_address.like("%@aol.com")).subquery()
address_subq = aliased(Address, subq)
stmt = (
    select(User, address_subq)
    .join_from(User, address_subq)
    .order_by(User.id, address_subq.id)
)
with Session(engine) as session:
    for user, address in session.execute(stmt):
        print(f"{user} {address}")
        
        
# CTE构造
cte_obj = select(Address).where(~Address.email_address.like("%@aol.com")).cte()
address_cte = aliased(Address, cte_obj)
stmt = (
    select(User, address_cte)
    .join_from(User, address_cte)
    .order_by(User.id, address_cte.id)
)
with Session(engine) as session:
    for user, address in session.execute(stmt):
        print(f"{user} {address}")

标量子查询

标量子查询:返回零行或一行且仅返回一列的子查询
SQLAlchemy 使用 ScalarSelect 构造表示标量子查询，它是 ColumnElement 表达式层次结构的一部分，而常规子查询由 Subquery 构造表示，它在 FromClause 层次结构中。
标量子查询通常（但不一定）与聚合函数一起使用，聚合函数在之前的 带有 GROUP BY / HAVING 的聚合函数 中介绍过。 标量子查询通过使用 Select.scalar_subquery() 方法显式指示，如下所示。 当自身进行字符串化时，它的默认字符串形式呈现为从两个表选择的普通 SELECT 语句

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
subq = (
    select(func.count(address_table.c.id))
    .where(user_table.c.id == address_table.c.user_id)
    .scalar_subquery()
)
print(subq)


stmt = select(user_table.c.name, subq.label("address_count"))
print(stmt)

stmt = (
    select(
        user_table.c.name,
        address_table.c.email_address,
        subq.label("address_count"),
    )
    .join_from(user_table, address_table)
    .order_by(user_table.c.id, address_table.c.id)
)
print(stmt)

# 为了指定 user_table 是我们要关联的表，我们使用 ScalarSelect.correlate() 或 ScalarSelect.correlate_except() 方法来指定
subq = (
    select(func.count(address_table.c.id))
    .where(user_table.c.id == address_table.c.user_id)
    .scalar_subquery()
    .correlate(user_table)
)

# 该语句可以像任何其他语句一样返回此列的数据
with engine.connect() as conn:
    result = conn.execute(
        select(
            user_table.c.name,
            address_table.c.email_address,
            subq.label("address_count"),
        )
        .join_from(user_table, address_table)
        .order_by(user_table.c.id, address_table.c.id)
    )
    print(result.all())

集合操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
union()
union_all()
intersect()
intersect_all()
except_()
except_all()
# 这些函数都接受任意数量的子可选择项，这些子可选择项通常是 Select 构造，但也可能是现有的组合。

from sqlalchemy import union_all
stmt1 = select(user_table).where(user_table.c.name == "sandy")
stmt2 = select(user_table).where(user_table.c.name == "spongebob")
u = union_all(stmt1, stmt2)
with engine.connect() as conn:
    result = conn.execute(u)
    print(result.all())
    
    
# 要像 Select 一样将 CompoundSelect 用作子查询，它提供了 SelectBase.subquery() 方法，该方法将生成一个 Subquery 对象，其中包含一个 FromClause.c 集合，可以在封闭的 select() 中引用。
u_subq = u.subquery()
stmt = (
    select(u_subq.c.name, address_table.c.email_address)
    .join_from(address_table, u_subq)
    .order_by(u_subq.c.name, address_table.c.email_address)
)
with engine.connect() as conn:
    result = conn.execute(stmt)
    print(result.all())
    
# orm实体
# 对于不嵌套在子查询中的简单 SELECT 与 UNION，通常可以通过使用 Select.from_statement() 方法在 ORM 对象获取上下文中使用它们。使用此方法，UNION 语句表示整个查询；在使用 Select.from_statement() 后，不能添加其他条件。
stmt1 = select(User).where(User.name == "sandy")
stmt2 = select(User).where(User.name == "spongebob")
u = union_all(stmt1, stmt2)

orm_stmt = select(User).from_statement(u)
with Session(engine) as session:
    for obj in session.execute(orm_stmt).scalars():
        print(obj)
        
        
user_alias = aliased(User, u.subquery())
orm_stmt = select(user_alias).order_by(user_alias.id)
with Session(engine) as session:
    for obj in session.execute(orm_stmt).scalars():
        print(obj)

exists子查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
subq = (
    select(func.count(address_table.c.id))
    .where(user_table.c.id == address_table.c.user_id)
    .group_by(address_table.c.user_id)
    .having(func.count(address_table.c.id) > 1)
).exists()
with engine.connect() as conn:
    result = conn.execute(select(user_table.c.name).where(subq))
    print(result.all())
    
# not exists: ~
subq = (
    select(address_table.c.id).where(user_table.c.id == address_table.c.user_id)
).exists()
with engine.connect() as conn:
    result = conn.execute(select(user_table.c.name).where(~subq))
    print(result.all())

sql函数

1
2
3
4
5
6
7
print(select(func.count()).select_from(user_table))
print(select(func.lower("A String With Much UPPERCASE")))
stmt = select(func.now())
with engine.connect() as conn:
    result = conn.execute(stmt)
    print(result.all())
print(select(func.some_crazy_function(user_table.c.name, 17)))

返回类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# pre-configured SQL function (only a few dozen of these)
func.now().type

# arbitrary SQL function (all other SQL functions)
func.run_some_calculation().type


from sqlalchemy import JSON
function_expr = func.json_object('{a, 1, b, "def", c, 3.5}', type_=JSON)


m1 = func.max(Column("some_int", Integer))
m1.type

m2 = func.max(Column("some_str", String))
m2.type


func.now().type # DateTime()
func.current_date().type # Date()

update

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
from sqlalchemy import update
stmt = (
    update(user_table)
    .where(user_table.c.name == "patrick")
    .values(fullname="Patrick the Star")
)
print(stmt)

stmt = update(user_table).values(fullname="Username: " + user_table.c.name)
print(stmt)


from sqlalchemy import bindparam
stmt = (
    update(user_table)
    .where(user_table.c.name == bindparam("oldname"))
    .values(name=bindparam("newname"))
)
with engine.begin() as conn:
    conn.execute(
        stmt,
        [
            {"oldname": "jack", "newname": "ed"},
            {"oldname": "wendy", "newname": "mary"},
            {"oldname": "jim", "newname": "jake"},
        ],
    )
    
# 相关更新
scalar_subq = (
    select(address_table.c.email_address)
    .where(address_table.c.user_id == user_table.c.id)
    .order_by(address_table.c.id)
    .limit(1)
    .scalar_subquery()
)
update_stmt = update(user_table).values(fullname=scalar_subq)
print(update_stmt)

# UPDATE...FROM
update_stmt = (
    update(user_table)
    .where(user_table.c.id == address_table.c.user_id)
    .where(address_table.c.email_address == "patrick@aol.com")
    .values(fullname="Pat")
)
print(update_stmt)

# MYSQL特定语法,可以更新多个表
update_stmt = (
    update(user_table)
    .where(user_table.c.id == address_table.c.user_id)
    .where(address_table.c.email_address == "patrick@aol.com")
    .values(
        {
            user_table.c.fullname: "Pat",
            address_table.c.email_address: "pat@aol.com",
        }
    )
)

# Update.ordered_values() 方法接受元组序列，以便可以控制此顺序
from sqlalchemy.dialects import mysql
print(update_stmt.compile(dialect=mysql.dialect()))

update_stmt = update(some_table).ordered_values(
    (some_table.c.y, 20), (some_table.c.x, some_table.c.y + 10)
)
print(update_stmt)

delete

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from sqlalchemy import delete
stmt = delete(user_table).where(user_table.c.name == "patrick")
print(stmt)

delete_stmt = (
    delete(user_table)
    .where(user_table.c.id == address_table.c.user_id)
    .where(address_table.c.email_address == "patrick@aol.com")
)
from sqlalchemy.dialects import mysql
print(delete_stmt.compile(dialect=mysql.dialect()))


# 从 UPDATE、DELETE 获取受影响的行数
with engine.begin() as conn:
    result = conn.execute(
        update(user_table)
        .values(fullname="Patrick McStar")
        .where(user_table.c.name == "patrick")
    )
    print(result.rowcount)
    
# 将 RETURNING 与 UPDATE、DELETE 一起使用
update_stmt = (
    update(user_table)
    .where(user_table.c.name == "patrick")
    .values(fullname="Patrick the Star")
    .returning(user_table.c.id, user_table.c.name)
)
print(update_stmt)
delete_stmt = (
    delete(user_table)
    .where(user_table.c.name == "patrick")
    .returning(user_table.c.id, user_table.c.name)
)
print(delete_stmt)

peewee

peewee：轻量 python orm 库，易于学习，直观使用。

  • Python 中是最流行的 ORM 框架之一，以简单为卖点
  • 同时兼容多种数据库系统：如 sqlite, mysql、postgresql
    • sqlite 扩展：包括 sqlite 日期操作函数的 cython 实现、regexp 运算符和全文搜索结果排名算法
    • pymysql：mysql
    • psycopg2：postgres
  • 功能：
    • model definition
    • storing data
    • retrieving data 数据检索
  • 参考资料：

基本概念

  • Model class 类 = 数据库的表
  • Field instance 实例 = Column 表的列
  • Model instance 实例 = Row 表的行

基本使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
# 直接连接
import peewee
db = peewee.MySQLDatabase('test', user='chen', password='chen', host='localhost', charset='utf8mb4')


# 连接池形式
from playhouse.pool import PooledMySQLDatabase
db = PooledMySQLDatabase('test', user='chen', password='chen', host='localhost', charset='utf8mb4')

# 公共的Model，存储公共的字段和方法
class BaseModel(Model):
    create_time = DateTimeField(default=datetime.datetime.now)
    update_time = DateTimeField(default=datetime.datetime.now)
    deleted     = IntegerField()

    def save(self, *args, **kwargs):
        self.update_time = datetime.datetime.now()
        return super(BaseModel, self).save(*args, **kwargs)

    class Meta:
        database = db

# 即使表将包含多个用户，我们总是使用单数形式命名类。
class User(BaseModel):
    class Meta:
        table_name = 'user'


    id = AutoField() # 自增的主键
    username = CharField(max_length=32) # varchar
    usertype=ForeignKeyField(Person, backref='pets')
    name = CharField(verbose_name='姓名', max_length=10, null=False, index=True)
    gender = IntegerField(verbose_name='姓别', null=False, default=1)
    birthday = DateField(verbose_name='生日', null=True, default=None)

# 创建表
db.create_tables([User]) # 默认使用 if not exists


# 处理完毕，关闭连接
db.close()
db.is_closed()

db.bind([User,Person]) # 将模型绑定到数据库
1
2
3
4
5
6
# SQL
# Bad! DO NOT DO THIS!
query = MyModel.select().where(SQL('Some SQL expression %s' % user_data))

# Good. `user_data` will be treated as a parameter.
query = MyModel.select().where(SQL('Some SQL expression %s', user_data))

连接状态

1
2
3
4
5
python
db.connect()

if not db.is_closed():
 db.close()

字段类型

  • 避免字段命名冲突，如 命名为 create
  • 查看表的所有列名：_._meta.fields.keys()
Field TypeSqlitePostgresqlMySQL
AutoFieldintegerserialinteger
BigAutoFieldintegerbigserialbigint
IntegerFieldintegerintegerinteger
BigIntegerFieldintegerbigintbigint
SmallIntegerFieldintegersmallintsmallint
IdentityFieldnot supportedint identitynot supported
FloatFieldrealrealreal
DoubleFieldrealdouble precisiondouble precision
DecimalFielddecimalnumericnumeric
CharFieldvarcharvarcharvarchar
FixedCharFieldcharcharchar
TextFieldtexttexttext
BlobFieldblobbyteablob
BitFieldintegerbigintbigint
BigBitFieldblobbyteablob
UUIDFieldtextuuidvarchar(40)
BinaryUUIDFieldblobbyteavarbinary(16)
DateTimeFielddatetimetimestampdatetime
DateFielddatedatedate
TimeFieldtimetimetime
TimestampFieldintegerintegerinteger
IPFieldintegerbigintbigint
BooleanFieldintegerbooleanbool
BareFielduntypednot supportednot supported
ForeignKeyFieldintegerintegerinteger
字段类型特殊参数
CharFieldmax_length，最大长度，默认为 255
FixedCharFieldmax_length
DateTimeFieldformats，'%Y-%m-%d %H:%M:%S' 具有属性 year, month, day, hour, minute, second
DateFieldformats，具有属性 year, month, day
TimeFieldformats，具有属性 hour, minute, second
TimestampFieldresolution, utc
DecimalFieldmax_digits=15 最大位数, decimal_places=2 最大精度, auto_round=True 自动舍入值, rounding
ForeignKeyField
BareFieldadapt
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 参数与默认值

null = False # 是否允许空字符串
index = False # 是否在此列上创建索引
unique = False # 是否在此列上创建唯一索引
column_name = None # 在数据库中显式指定列名，不设置会使用定义model时的变量名作为列名。
default = None  # 默认值,任何值或可作为未初始化模型的默认值使用的调用
primary_key = False # 表的主键
constraints = None # 增加约束列表，如：[Check('price > 0')]
sequence = None # 序列名（如果后端支持）
collation = None # 用于排序字段/索引的排序规则
unindexed = False # 指示应取消对虚拟表上的字段的索引（sqlite only*）
choices = None # 设置可选的枚举选项，值为一个元组，元素也是一个元素，如((1,'女'),(2,'男'))。
help_text = None # 设置备注或注释文本
verbose_name = None # 表示此字段的“用户友好”的名称
index_type = None # 指定自定义索引类型，一般默认即可。例如，对于Postgres，可以指定 'BRIN' 或 'GIN' 索引。

常用字段类型的特殊参数

1
2
3
4
5
CharField、FixedCharField：
max_length = 32 设置给字段最大长度为32，注意CharField的最大max_length=255。
DateTimeField、DateField、TimeField：
formats = '%Y-%m-%d %H:%M:%S' 设置日期时间的格式化，最常用的格式化。
formats = '%Y-%m-%d' 设置日期的格式化，最常用的格式化。
1
2
3
4
DecimalField：
max_digits = 15 设置要存储的最大位数。
decimal_places = 2 设置最大精度。
auto_round = True 设置自动舍入值。
1
2
3
4
5
6
7
8
9
10
11
ForeignKeyField：
model = Model object 这是要引用的model或者外键指向的model，必填。
field = Field 设置外键指向model的具体字段，默认是主键，无特殊需求不用更改，默认即可。
backref = 'str' 设置反向引用的访问器名称，类似于Django中设置外键字段是配置的related_name。
on_delete = 'CASCADE' 设置当被引用的记录删除时，指向该记录的数据操作，存在多个可选值：
 CASCADE 跟随外键一同删除
 RESTRICT 限制外表中的数据删除，就是当外键删除时，检测到有指向该外键的数据，那么就不允许删除。
 SET NULL 设置空值
 SET DEFAULT 设置默认值
 NO ACTION 不做操作，默认的
on_update = 'CASCADE' 和on_delete作用类似，只是on_update实在外键数据修改时触发。

主键索引约束

期权意义Inheritable？
database模型数据库对
table_name要存储数据的表的名称不
table_function动态生成表名的函数对
indexes要索引的字段列表对
primary_key一 CompositeKey 实例对
constraints表约束列表对
schema模型的数据库架构对
only_save_dirty调用 model.save（）时，只保存脏字段对
options用于创建表扩展名的选项字典对
table_settings将字符串设置在右括号后的列表对
temporary表示临时表对
legacy_table_names使用旧表名生成（默认情况下启用）对
depends_on指示此表依赖另一个表进行创建不
without_rowid指示表不应具有 rowid（仅限于 sqlite）不
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
# 主键
class BlogToTag(Model):
    # 直接创建AutoField，会默认做为自动递增的整数主键
    id = AutoField()
    """A simple "through" table for many-to-many relationship."""
    blog = ForeignKeyField(Blog)
    tag = ForeignKeyField(Tag)

    class Meta:
        # 复合主键
        primary_key = CompositeKey('blog', 'tag')

class NoPrimaryKey(Model):
    data = IntegerField()

    class Meta:
        # 设置为 False，无主键
        primary_key = False


# 外键
import datetime
from peewee import *


db = SqliteDatabase(':memory:')

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

class User(BaseModel):
    username = TextField()

class Tweet(BaseModel):
    content = TextField()
    timestamp = DateTimeField(default=datetime.datetime.now)
    user = ForeignKeyField(User, backref='tweets')
`model`外键指向的model, `field`外键model的具体字段，默认为主键, `backref`设置反向引用的访问器, `on_delete`, `on_update`, `deferrable` `lazy_load`
on_delete = 'CASCADE' 设置当被引用的记录删除时，指向该记录的数据操作，存在多个可选值：
 CASCADE 跟随外键一同删除
 RESTRICT 限制外表中的数据删除，就是当外键删除时，检测到有指向该外键的数据，那么就不允许删除。
 SET NULL 设置空值
 SET DEFAULT 设置默认值
 NO ACTION 不做操作，默认的
on_update = 'CASCADE' 和on_delete作用类似，只是on_update实在外键数据修改时触发。


class Favorite(BaseModel):
    user = ForeignKeyField(User, backref='favorites')
    tweet = ForeignKeyField(Tweet, backref='favorites')


class User(Model):
    # unique约束
    username = CharField(unique=True)
    # 索引
    email = CharField(index=True)

# 约束
class Product(Model):
    name = CharField(unique=True)
    price = DecimalField(constraints=[Check('price < 10000')])
    created = DateTimeField(
        constraints=[SQL("DEFAULT (datetime('now'))")])

    class Meta:
        constraints=[Check('price < 10000')]
# 多列索引：定义为 Meta 使用嵌套元组的属性。每个数据库索引都是一个2元组，第一部分是字段名称的元组，第二部分是指示索引是否应唯一的布尔值。
class Transaction(Model):
    from_acct = CharField()
    to_acct = CharField()
    amount = DecimalField()
    date = DateTimeField()

    class Meta:
        indexes = (
            # create a unique on from/to/date
            (('from_acct', 'to_acct', 'date'), True),

            # create a non-unique on from/to
            (('from_acct', 'to_acct'), False),
        )
#  Model.add_index()
# Add an index on "name" and "timestamp" columns.
Article.add_index(Article.name, Article.timestamp)

# Add a partial index on name and timestamp where status = 1.
Article.add_index(Article.name, Article.timestamp,
                  where=(Article.status == 1))

增 -insert

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
# Model.save()
# 需创建实例
user_chen = User(username='chen',age=18)
user_chen.name = 'chen2' # 这种方式可随时改
user_chen.save()

# Model.create()
data_list=[{'name':'chen'},{'name':'zhang'},{'name':'li'}]
in_id=User.insert(data_list).execute()

# Model.insert()
user_id = User.insert(name='张三',age=26,sex='男').execute() # 创建一个记录，并返回新纪录的主键 这里使用了 execute() 方法，因为insert不会马上执行，execute()方法可以马上执行操作。


# insert_many
User.insert_many([
    ('张三丰', 26, '男'),
    ('王大胖', 12, '男'),
    ('王小小', 22, '女')
],
    ['name', 'age', 'sex']
).execute()
也可以插入字典列表
User.insert_many([
    {'name': 王二小', 'age': 12, 'sex': '男'},
    {'name': 王小小', 'age': 13, 'sex': '女'},
    {'name': 王大胖', 'age': 15, 'sex': '男'}
]).execute()

# ##### bulk_create
和Django类似的批量创建
users = [
    {'name': 王二小', 'age': 12, 'sex': '男'},
    {'name': 王小小', 'age': 13, 'sex': '女'},
    {'name': 王大胖', 'age': 15, 'sex': '男'}
]
datas = [User(**item) for item in users]
# 不使用事务
User.bulk_create(datas) 
# 使用事务
with db.atomic():
    User.bulk_create(datas)

# 只希望插入数据而不需要创建模型实例时使用
User.insert(username='chen', age=18).execute()


# 插入多条，不要用for一条条来，insert_many已内部优化，更快
data_list = [{'username': 'u1'}, {'username': 'u2'}, {'username': 'u3'}]
User.insert_many(data_list).execute()

# 批量插入插入大批量数据data，已经按照上面的格式构造好，设置一次插入100行.
from peewee import chunked

with db.atomic():
 for batch in chunked(data_list, 100):
     User.insert_many(batch).execute()

# 从另一个表插入
res = (TweetArchive
       .insert_from(
           Tweet.select(Tweet.user, Tweet.message),
           fields=[TweetArchive.user, TweetArchive.message])
       .execute())
# 等价于
INSERT INTO "tweet_archive" ("user_id", "message")
SELECT "user_id", "message" FROM "tweet";

删

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 删, delete: Model.delete_instance() 和 Model.delete()
User.delete().where(User.id==1).execute()

p = Person(name='tom', gender=1, birthday=date(2000, 1, 1))
p.save()

p.delete_instance()

# delete()
User.delete().where(User.id == 1).execute() # 可以应用批量删除
User.delete().where(User.age < 18).execute() # 可以删除所有年龄小于18的数据行


# detete_instance
user = User.get(User.id=1) # 针对已经知道明确的对象时，可以应用该删除方法更好些，可以避免误删。
user.delete_instance()

使用事务：使用事务可以保证原子性等事务的四大特性
with db.atomic():
    # 执行相关SQL操作
    user = User.get(User.id=1)
    user.age = 18
    user.save()

改

1
2
3
4
5
6
7
8
9
10
# Model.save() 
now_user.name='chen'
now_user.save()


# Model.update()
# 改
User.update({'username':'zhang'}).where(User.id ==1).execute() # 优先这种，Python字典风
User.update({User.username='zhang'}).where(User.id ==1).execute()

查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
# Model.get()
# 查询得到单条数据
User.get(User.id == 1)
# 避免查询不到时报错，返回None
User.get_or_none(User.id == 1)

User.get_by_id(1)  # Same as above.
User[1]  # Also same as above.

# get_or_creat
user, created = User.get_or_create(name='张三', defaults={'age': 26, 'sex': '男'}) # 如果存在就返回user，不存在就创建后再返回user
# Model.select
# 多条数据，加 where 限制条件，不加则查询所有数据
user = User.select() # 查询user内的所有数据，查询所有列
user = User.select(User.id, User.name) # 查询列为id、name，返回所有数据
user = User.select(User.id, User.name).where(User.name=='chen')
for _ in user:
    print(user.id,user.name)


user = User.get(User.id==1) # 当获取的结果不存在时，报 Model.DoesNotExist 异常。如果有多条记录满足条件，返回第一条。
user = User.get_or_none(User.id==1) # 和get的区别是，当没有匹配数据时返回None，不会抛出异常。
user = User.get_by_id(1) # 对于主键查找可以用这个
user, created = User.get_or_create(name='张三', defaults={
    'age': 26, 'sex': '男'}) # 如果存在就返回user，不存在就创建后再返回user


带where条件的：
where语句中查询运算符：
## 与或非逻辑运算符：
& 与 示例：(User.is_active == True) & (User.is_admin == True)
|  或 示例：(User.is_admin) | (User.is_superuser)
~ 非 示例：~(User.username.contains('admin'))
比较运算符：
== 等于
< 小于
<= 小等于
> 大于
>= 大等于
!= 不等于
<< x IN y，其中y是列表或查询
>> x IS y，其中y可以是None / NULL，例如判断obj_id >> None 相当于SQL语句：obj_id is NULL
% x喜欢y，其中y可能包含通配符
** x像y，其中y可能包含通配符
^ 异或
~ 一元否定（例如，NOT x）
## 和MySQL查询的一些运算符：
.in_(value)  IN查找（与相同<<）。
.not_in(value) 不在查询中。
.is_null(is_null) IS NULL或IS NOT NULL。接受布尔参数。
.contains(substr) 通配符搜索子字符串。相当于 LIKE '%xx%'
.startswith(prefix) 搜索以开头的值prefix。
.endswith(suffix) 搜索以结尾的值suffix。
.between(low, high) 在low和之间搜索值high。
.regexp(exp) 正则表达式匹配（区分大小写）。
.iregexp(exp) 正则表达式匹配（不区分大小写）。
.bin_and(value) 二进制AND。
.bin_or(value) 二进制或。
.concat(other) 使用串联两个字符串或对象||。
.distinct()  标记列以进行DISTINCT选择。
.collate(collation) 用给定的排序规则指定列。
.cast(type)  将列的值强制转换为给定的类型。

示例：
user = User.select().where(User.name == '张三') # 查询所有叫张三的数据
user = User.select().where(User.name == '张三', User.age = 26) # 查询名字叫张三并且年龄为26的数据
user = User.select().where((User.name == '张三') | (User.age = 26)) # 查询名字叫张三或者年龄为26的数据
user = User.select().where(~User.age == 26)# 查询年龄不等于26的数据
user = User.select().where(User.age != 26)# 查询年龄不等于26的数据
user = User.select().where(User.age >= 26)# 查询年龄大等于26的数据
user = User.select().where(User.age in [24,26,27]) # 查询年龄在[24,26,27]内的数据
user = User.select().where(User.name.contains('张三')) # 模糊查询，相当于 LIKE '%张三%'
排序：
user = User.select().order_by(User.id) # 默认升序 == order_by(User.id.asc())
user = User.select().order_by(-User.id) # 降序排序 == order_by(User.id.desc())
取出前N个数据：
user = User.select().order_by(-User.id).limit(10) # 取出id从大到小的前10个数据
计数count：
tatol = User.select().count() # 统计所有数据条数
分页查询：
user = User.select().order_by(User.id).paginate(2, 10) # paginate(page, page_size) 查出第二页，每页10条数据
分组聚合查询：
query = User.select(User.age, fn.Count(User.id).alias('count')).group_by(User.age)
假设现在有三个表分别为：
class Photo(Model):
    image = CharField()

class Tag(Model):
    name = CharField()

class PhotoTag(Model):
    photo = ForeignKeyField(Photo)
    tag = ForeignKeyField(Tag)

query = (Tag.select().join(PhotoTag).join(Photo).group_by(Tag).having(fn.Count(Photo.id) > 5))


select_queries = User.select(...).where(...).order_by(...).limit(...)...
chen=Person.select().where(Person.name=='chen').get()
chen=Person.get(Person.name=='chen')
print(chen.name,chen.age)

chens = User.select().where(User.username == 'chen')
for user in chens:
    print(user.create_time,user.username)

# raw作为保留使用，是为了显著优化查询的场景使用。因为它返回模型的实例所以select插询很有用。
q = User.raw('select id, username from users')
for user in q:
    print user.id, user.username


# 关联查询
for pet in Pet.select().join(Person).where(Person.name == 'Bob'):
    print(pet.name)
query = (Tweet
         .select()
         .join(User)
         .order_by(User.username, Tweet.created_date.desc()))

SELECT t1."id", t1."user_id", t1."message", t1."is_published", t1."created_date"
FROM "tweet" AS t1
INNER JOIN "user" AS t2
  ON t1."user_id" = t2."id"
ORDER BY t2."username", t1."created_date" DESC

# 创建或获取，它首先尝试检索匹配的行。如果失败，将创建一个新行。
person, created = Person.get_or_create(
    first_name=first_name,
    last_name=last_name,
    defaults={'dob': dob, 'favorite_color': 'green'})

# 减少查询所使用的内存量，iterator()
# Let's assume we've got 10 million stat objects to dump to a csv file.
stats = Stat.select()

# Our imaginary serializer class
serializer = CSVSerializer()

# Loop over all the stats and serialize.
for stat in stats.iterator():
# for stat in stats.dicts.iterator():
# for stat in stats.namedtuples.iterator():
# for stat in stats.tuples.iterator():
    serializer.serialize_object(stat)

# 获取随机记录
LotteryNumber.select().order_by(fn.Rand()).limit(5)
# 排序
for t in Tweet.select().order_by(Tweet.created_date.desc()):

query = (User
             .select(User.username, fn.COUNT(Tweet.id).alias('num_tweets'))
             .join(Tweet, JOIN.LEFT_OUTER)
             .group_by(User.username)
             .order_by(fn.COUNT(Tweet.id).desc()))
# 获取分页记录
for tweet in Tweet.select().order_by(Tweet.id).paginate(2, 10):
 print(tweet.message)
# 统计行数
Tweet.select().where(Tweet.id > 50).count()
# 聚合记录
query = (Tag
         .select(Tag, fn.Count(Photo.id).alias('count'))
         .join(PhotoTag)
         .join(Photo)
         .group_by(Tag)
         .having(fn.Count(Photo.id) > 5))

# 检索标量值
PageView.select(fn.Count(fn.Distinct(PageView.url))).scalar()
  • 请注意，peewee 使用 bitwise 操作员（ & 和 | ）而不是逻辑运算符（ and 和 or ）原因是 python 将逻辑操作的返回值强制为布尔值。这也是必须使用 .in_() 而不是 in 操作员。
    • 使用 .in_() 和 .not_in() 而不是 in 和 not in
    • 使用 & 而不是 and
    • 使用 | 而不是 or
    • 使用 ~ 而不是 not
    • 使用 .is_null() 而不是 is None 或 == None .
    • 使用逻辑运算符时，不要忘记将比较结果括在括号中。
1
2
# where
for tweet in Tweet.select().where(Tweet.created_date < datetime.datetime(2011, 1, 1)):
比较意义
==x 等于 y
<x 小于 y
<=x 小于或等于 y
>x 大于 y
>=x 大于或等于 y
!=x 不等于 y
<<x 代表 y，其中 y 代表列表或查询
>>x 是 y，其中 y 是无/空
%x 类似 y，其中 y 可能包含通配符
**x i 类似 y，其中 y 可能包含通配符
^X 异或 Y
~一元否定（例如，不是 x）
方法意义
.in_(value)在查找中（与 << ）
.not_in(value)不在查找中。
.is_null(is_null)为空或不为空。接受布尔参数。
.contains(substr)子字符串的通配符搜索。
.startswith(prefix)搜索以开头的值 prefix .
.endswith(suffix)搜索以结尾的值 suffix .
.between(low, high)搜索介于 low 和 high .
.regexp(exp)正则表达式匹配（区分大小写）。
.iregexp(exp)正则表达式匹配（不区分大小写）。
.bin_and(value)二进制和
.bin_or(value)二进制或
.concat(other)使用连接两个字符串或对象 `` .
.distinct()为非重复选择标记列。
.collate(collation)使用给定的排序规则指定列。
.cast(type)将列的值强制转换为给定的类型。

高级与常用

model_to_dict：将查询到的结果转为 dict，backrefs 是一个千金这项，表示返回关联表的数据
json_serial：转换 datetime 为 json 格式的方法，可以避免包含 datetime 时转换出错

遇到的问题

  • 问题：peewee.ImproperlyConfigured: MySQL driver not installed!
  • 解决方案：
1
pip install pymysql

dataset

dataset：提供了一个简单的抽象层， 无需完整的 ORM 可省大量 SQL 语句，可像 JSON 和 NoSQL 一样操作数据库

1
2
3
4
5
6
import dataset
db=dataset.connect('sqlit://:memory:')
table=db['table1']
table.insert(dict(name='Chen',age=18))
table.insert(dict(name='zhang',age=20,gender='female'))
chen=table.find_one(name='Chen')
  • 多线程连接 sqlite3 时，报：SQLite objects created in a thread can only be used in that same thread：
1
2
# 连接配置中加入；
check_same_thread=Fasle

Gadfly

Gadfly：提供了一个完整的 SQL 环境

sandman2

sandman2：号称不用一行代码，用 API 操作数据库，不咋地，OUT

  • 基于已存在的数据库，生成一个 RESTful API 服务器
  • python3.8+Windows 上有个小 BUG，添加数据时报 If this is an auto-generated value, check that the database table allows generation of new primary key values...，界面不美观，OUT
相关文章
给博主来一杯卡布奇诺