Chiriri's blog
Iekr
2022-04-22
目录

SQLalchemy

# SQLalchemy

sqlalchemy (opens new window) 是 Python 编程语言下的一款 ORM 框架，该框架建立在数据库 API 之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成 SQL，然后使用数据 API 执行 SQL 并获取执行结果。

# 创建连接

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

# 创建对象的基类
Base = declarative_base()
# 初始化数据库连接
engine = create_engine(
    #'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
    "mysql+pymysql://root:123456@localhost:3306/pyorm",
    # 对数据库进行编码设置，能对数据库进行中文读写
    charset="utf8",
    # pool_size用于设置连接数，默认设置5个
    pool_size=5,
    # erflow默认连接数为10。当超出最大连接数后，如果超出的连接数在max_overflow设置的访问内，超出部分可以继续连接访问，
    max_overflow=4,
    # pool_recycle: 连接重置周期，默认为-1，推荐设置为7200，即如果连接已空闲7200秒，就自动重新获取，防止connection被关闭
    pool_recycle=7200,
    # pool_timeout:连接超时时间，默认为30秒，超过时间的连接会连接失败
    pool_timeout=30)
# 创建DBSession类型
DBSession = sessionmaker(bind=engine)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
  • echo=False -- 如果为真，引擎将记录所有语句以及 repr() 其参数列表的默认日志处理程序。
  • enable_from_linting -- 默认为 True。如果发现给定的 SELECT 语句与将导致笛卡尔积的元素取消链接，则将发出警告。
  • encoding -- 默认为 utf-8
  • future -- 使用 2.0 样式
  • hide_parameters -- 布尔值，当设置为 True 时，SQL 语句参数将不会显示在信息日志中，也不会格式化为 StatementError 对象。
  • listeners -- 一个或多个列表 PoolListener 将接收连接池事件的对象。
  • logging_name -- 字符串标识符，默认为对象 id 的十六进制字符串。
  • max_identifier_length -- 整数；重写方言确定的最大标识符长度。
  • max_overflow=10 -- 允许在连接池中 “溢出” 的连接数，即可以在池大小设置（默认为 5）之上或之外打开的连接数。
  • pool_size=5 -- 在连接池中保持打开的连接数
  • plugins -- 要加载的插件名称的字符串列表。

# 定义映射对象

必须先初始化连接

tablename 代表表名

Column : 代表数据表中的一列，内部定义了数据类型

**primary_key：** 主键

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index

# 定义User对象:
class User(Base):
    # 表的名字:
    __tablename__ = 'user'

    # 表的结构:
    id = Column(String(20), primary_key=True)
    name = Column(String(20))
1
2
3
4
5
6
7
8
9
10

primary_key=True 主键索引 autoincrement=True 自增字段 index=True 给当前字段创建普通索引 unique=True 给当前字段创建唯一索引 UniqueConstraint (' 字段 ',' 字段 ',name=' 索引名称 ') 创建唯一组合索引 Index (' 索引名称 ',' 字段 ',' 字段 ') 创建普通组合索引 default='abc' 设置字段默认值，不怎么可靠 ForeignKey ("连接表名称。连接表主键字段") 设置外键链表 nullable=False 类容不能为空

整数型：TINYINT，SMALLINT，INT，BIGINT

  1. Boolean () 对应 TINYINT
  2. Integer () 对应 INT
  3. SMALLINT () 对应 SMALLINT
  4. BIGINT () 对应 BIGINT

浮点型：FLOAT，DOUBLE，DECIMAL (M,D)

  1. DECIMAL () 对应 DECIMAL
  2. Float () 对应 FLOAT
  3. REAL () 对应 DOUBLE

字符型：CHAR，VARCHAR

  1. String (40) 对应 VARCHAR
  2. CHAR () 对应 CHAR

日期型：DATETIME，DATE，TIMESTAMP

  1. DATETIME () 对应 DATETIME
  2. DATE () 对应 DATE
  3. TIMESTAMP () 对应 TIMESTAMP

备注型：TINYTEXT，TEXT，

  1. Text () 对应 TEXT
  2. UnicodeText (10) 对应 TINYTEXT

# 一对多映射类

通过外键绑定主键

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'))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 多对多

class Server(Base):
    __tablename__ = "server"
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(8))


class Group(Base):
    __tablename__ = "group"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(16), unique=True)


class ServertoGroup(Base):
    __tablename__ = "servertogroup"
    id = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 创建表

必须创建映射类并指定表名 才会创建 如表存在不进行操作 如无则创建

class Lagoujob(Base):
    # 指定表名
    __tablename__='lagoujob'
    id=Column(Integer,primary_key=True)
    area=Column(String(50))
    community=Column(String(50))
    village=Column(String(50))


engine = create_engine("mysql+pymysql://root:123456@localhost:3306/pyorm", max_overflow=4)
Base.metadata.create_all(engine)
1
2
3
4
5
6
7
8
9
10
11

# 增删改查

# 增

# 创建session对象:
session = DBSession()
# 创建新User对象:
new_user = User(id='5', name='Bob')
# 添加到session:
session.add(new_user)
# 提交即保存到数据库:
session.commit()
# 关闭session:
session.close()
1
2
3
4
5
6
7
8
9
10

# 删

user = session.query(User).filter(User.id=='5').first()
session.delete(user)
session.commit()
1
2
3

# 改

emp = session.query(Employee).filter_by(EMP_ID='9002').first()
emp.AGE = '21'
session.commit()

#写法二
session.query(table1).filter(table1.id > 8).update({table1.age: 11})
1
2
3
4
5
6

# 查

  • query.filter() 过滤
  • query.filter_by() 根据关键字过滤
  • query.all() 返回列表
  • query.first() 返回第一个元素
  • query.one() 有且只有一个元素时才正确返回
  • query.one_or_none()，类似 one，但如果没有找到结果，则不会引发错误
  • query.scalar()，调用 one 方法，并在成功时返回行的第一列
  • query.count() 计数
  • query.order_by() 排序
# 创建Session:
session = DBSession()

# 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
user = session.query(User).filter(User.id=='5').one()

# Query 对象提供了 filter() 方法和 filter_by() 方法用于数据筛选。filter_by() 适用于简单的基于关键字参数的筛选。 filter() 适用于复杂条件的表达。
user = session.query(User).filter_by(User.id=='5').first()

# 多条件查询 条件之间用逗号隔开
get_data = session.query(table1).filter(table1.id > 1, table1.age > 5).all()

# 查询所有
all_user = session.query(User).all()

# 只获取指定字段 返回为由元组组成的list
all = session.query(user.id, user.name).all()

#输出查询结果 默认为list[user_Object]
print(all_user)

# 关闭Session:
session.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

  • query.join() 连接查询

    • join 分为 left join （左外连接）和 right join （右外连接）以及 内连接 （等值连接）。
    • 在 sqlalchemy 中，使用 join 来完成内连接。在写 join 的时候，如果不写 join 的条件，那么默认将使用 外键 来作为条件连接。
    • 查询出来的字段，跟 join 后面的东西无关，而是取决于 query 方法中传了什么参数。（模型名 = 全表；模型名。属性 = 表名。字段）。
    • 在 sqlalchemy 中，使用 outer join 来完成外连接（默认是左外连接）。
    session.query(BvVideo.ctime,UpAuthor.name).join(UpAuthor,BvVideo.mid == UpAuthor.up_mid).all()
    1

  • query(column.label()) 别名

    session.query(User.name.label('name_label')).all()
    1

  • **aliased ()** 为查询对象设置别名

    user_alias = aliased(User, name='user_alias')
    1

# 一些常用条件的 query 查询

#== <= >=
def test_filter_le(self):
    emps = session.query(Employee).filter(Employee.EMP_ID <= '1009').all()
    print(to_formatted_table(emps))

#!=
def test_filter_ne(self):
    emps = session.query(Employee).filter(Employee.EMP_ID != '1001').all()
    print(to_formatted_table(emps))

#like
def test_filter_like(self):
    emps = session.query(Employee).filter(Employee.EMP_ID.like('%9')).all() #不区分大小写
    print(to_formatted_table(emps))

#in
def test_filter_in(self):
    emps = session.query(Employee).filter(Employee.EDUCATION.in_(['Bachelor', 'Master'])).all()
    print(to_formatted_table(emps))

#not in
def test_filter_notin(self):
    emps = session.query(Employee).filter(~Employee.EDUCATION.in_(['Bachelor', 'Master'])).all()
    print(to_formatted_table(emps))

#is null
def test_filter_isnull(self):
    emps = session.query(Employee).filter(Employee.MARITAL_STAT == None).all()
    print(to_formatted_table(emps))

#is not null
def test_filter_isnotnull(self):
    emps = session.query(Employee).filter(Employee.MARITAL_STAT != None).all()
    print(to_formatted_table(emps))

#ans
def test_filter_and(self):
    emps = session.query(Employee).filter(Employee.GENDER=='Female', Employee.EDUCATION=='Bachelor').all()
    print(to_formatted_table(emps))

#and
def test_filter_and2(self):
    emps = session.query(Employee).filter(and_(Employee.GENDER=='Female', Employee.EDUCATION=='Bachelor')).all()
    print(to_formatted_table(emps))

#or
def test_filter_or(self):
    emps = session.query(Employee).filter(or_(Employee.MARITAL_STAT=='Single', Employee.NR_OF_CHILDREN==0)).all()
    print(to_formatted_table(emps))
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

# 书写 sql 语句

from sqlalchemy import text
session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(User.id).one()
1
2

# 回滚

在 commit() 之前，对实例对象的属性所做的更改，可以进行回滚，回到更改之前。

本质上只是把某一条数据（也就是映射类的实例）从内存中删除而已，并没有对数据库有任何操作。

session.rollback()
1
编辑 (opens new window)
上次更新: 2023/12/06, 01:31:48
Falsk
Scrapy

← Scrapy→

最近更新
01
k8s
06-06
02
进程与线程
03-04
03
计算机操作系统概述
02-26
更多文章>
Theme by Vdoing | Copyright © 2022-2025 Iekr | Blog