devbox
正经夜光杯
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

（二）SQLAlchemy引擎和模型类_sqlalchemy oceanbase

作者：正经夜光杯 | 2024-08-13 14:36:26

赞

踩

sqlalchemy oceanbase
  • SQLAlchemy官方文档： https://www.osgeo.cn/sqlalchemy/
  • SQLAlchemy1.4教程： https://www.osgeo.cn/sqlalchemy/tutorial/engine.html
  • 本人使用的SQLAlchemy是1.4.23.

目录

一： 建立连接：

# settings.py
DB_HOST = 'xx.xxx.xxx.xx'
DB_PORT = xxxx
DB_USER = 'xxxxxx'
DB_PASSWORD = 'xxxxx'
DB_NAME = 'xxxxx'
DB_URI = f'postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}'

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
from sqlalchemy import create_engine
import settings
# echo是输出sql语句到python终端打印出来
engine = create_engine(settings.DB_URI, echo=True)

  • 1
  • 2
  • 3
  • 4

二： 处理事务与DBAPI

1: 获取连接：

  • 获取连接时，需要一些系统资源，而断开连接，需要释放一些资源，因此我们需要使用上下文来管理连接对象connection。

    from sqlalchemy import create_engine
import settings

engine = create_engine(settings.DB_URI, echo=True)

with engine.connect() as conn:
    sql = "select * from customer;"
    result = conn.execute(sql)
    for customer in result:
        print(customer.name)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

2：持久化到数据库：

  • 案例：新建一个表，名为users，然后持久化到数据库。

  • 我们可以发现的是：使用engine.begin()，和使用engine.connect()，都会提交事务，持久化到数据库。

    from sqlalchemy import create_engine
import settings

engine = create_engine(settings.DB_URI, echo=True)

with engine.connect() as conn:
    sql = "create table users(name varchar(255), age int);"
    result = conn.execute(sql)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    from sqlalchemy import create_engine
import settings

engine = create_engine(settings.DB_URI, echo=True)

with engine.begin() as conn:
    sql = "create table users(name varchar(255), age int);"
    result = conn.execute(sql)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

3：返回对象

  • 手动插入两个记录，然后查询数据：

    from sqlalchemy import create_engine
import settings

engine = create_engine(settings.DB_URI, echo=True)

with engine.begin() as conn:
    sql = "select * from users;"
    result = conn.execute(sql)
    for res in result:
        print(res.name, res.age)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

  • 我们执行完这个语句返回的result是一个可迭代对象Result。

  • 如果我们对这个Result对象进行遍历，得到的是每个Row(行)对象。

  • 1： Result对象可以直接转换成列表套元组的格式：

    with engine.begin() as conn:
    sql = "select * from users;"
    result = conn.execute(sql)
    print(result.all())
    # [('liangshan', 23), ('niuniu', 24)]

    • 1
    • 2
    • 3
    • 4
    • 5

  • 2： Result对象可以直接按列拆开，例如现在有两列name, age：

    with engine.begin() as conn:
    sql = "select * from users;"
    result = conn.execute(sql)
    for name, age in result:
        print(name, age)
        # liangshan 23
        # niuniu 24

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

  • 3： 遍历每个Result对象得到的是每个Row对象（元组）。

    with engine.begin() as conn:
    sql = "select * from users;"
    result = conn.execute(sql)
    for row in result:
        print(row)
        # ('liangshan', 23)
        # ('niuniu', 24)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

  • 4： Row既然是元组，那么就可以使用下标取值，除此之外,这个Row对象还拥有动态属性。将值和列名进行了绑定。

    with engine.begin() as conn:
    sql = "select * from users;"
    result = conn.execute(sql)
    for row in result:
        print(row.name, row.age)

    • 1
    • 2
    • 3
    • 4
    • 5

  • 5： 当然，Result对象默认是转化成Row元组对象，如果想要转化成字典对象也是有办法的。

    • 本质是将Row,先转化成 MappingResult 对象，遍历这个对象会得到RowMapping对象。这个就是一个字典对象。
    with engine.begin() as conn:
    sql = "select * from users;"
    result = conn.execute(sql)
    for row_dict in result.mappings():
        print(row_dict['name'])
        print(row_dict['age'])

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

4： 参数绑定：

  • 当我们使用SQLAlchemy传递参数的时候，SQlAlchemy是如何拼接sql语句的呢？
with engine.begin() as conn:
    sql = "select name, age  from users where age = {age};"
    result = conn.execute(sql.format(age=23))
    for row_dict in result.mappings():
        print(row_dict['name'], row_dict['age'])
        # liangshan 23

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

5： 会话：

  • 测试运行与之前的不同：

    engine = create_engine(settings.DB_URI, echo=True)
sql = "select name, age  from users where age = 23;"
with Session(engine) as session:
    result = session.execute(sql)
    for row in result:
        print(row.name, row.age)
        # liangshan 23

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

  • 现在结果与之前的结果：

    BEGIN (implicit)
select name, age  from users where age = 23;
2021-08-19 16:23:14,786 INFO sqlalchemy.engine.Engine [raw sql] {}
liangshan 23
COMMIT

    • 1
    • 2
    • 3
    • 4
    • 5
    # 现在的执行结果：
 BEGIN (implicit)
select name, age  from users where age = 23;
2021-08-19 16:19:47,170 INFO sqlalchemy.engine.Engine [generated in 0.00019s] {}
liangshan 23
ROLLBACK

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

  • 不同点：

    • 1： session执行的是回滚操作，而之前是提交操作（会话不会主动提交）。
    • 2： session执行会有一个时间提示（session是跟踪connection的）。

三：使用数据库元数据：

1: 了解三种元数据：

  • MataData: 是一个集合，存放表名。
  • Table: 表
  • Column: 列

2: 简单的模型类

  • Table会分配给MetaData管理。
  • Column会分配给Table管理。
from sqlalchemy import create_engine
import settings
from sqlalchemy import Table, Column, Integer, String,MetaData

engine = create_engine(settings.DB_URI, echo=True)

metadata = MetaData()
user_table = Table(
    "users",
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('age', Integer)
)

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

3: 常见的约束：

  • 类型约束：Interger, Varchar(255),
  • 主键约束：primary_key=True
  • 非空约束：nullable = True

4: 迁移工具的底层实现

  • 核心： metadata.create_all(engine)
from sqlalchemy import create_engine
import settings
from sqlalchemy import Table, Column, Integer, String,MetaData

engine = create_engine(settings.DB_URI, echo=True)

metadata = MetaData()
user_table = Table(
    "users",
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('age', Integer)
)
# 执行DDL语句
metadata.create_all(engine)

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

  • 查看数据库执行过程：

    该过程我们发现会先向数据库查询这个表是否存在，然后再创建。

BEGIN (implicit)
select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
[generated in 0.00016s] {'name': 'users'}
CREATE TABLE users (
	id SERIAL NOT NULL, 
	name VARCHAR(255), 
	age INTEGER, 
	PRIMARY KEY (id)
)
[no key 0.00019s] {}
COMMIT

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 同样我们也可以删除表：
from sqlalchemy import create_engine
import settings
from sqlalchemy import Table, Column, Integer, String,MetaData

engine = create_engine(settings.DB_URI, echo=True)

metadata = MetaData()
user_table = Table(
    "users",
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('age', Integer)
)
# 执行DDL语句，删除所管理的表
metadata.drop_all(engine)

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

5： 使用ORM定义元数据：

5.1: ORM中的注册表

  • 注册表封装了我们上面提到的metadata这个元数据。

  • 注册表的作用：

    • 1： 产生基类（后面说）
    • 2： 控制表的产生和销毁（metadata的作用）
    from sqlalchemy.orm import registry
from sqlalchemy import create_engine
import settings

engine = create_engine(settings.DB_URI, echo=True)
# 1：实例化注册表
mapper_register = registry()
# 2：打印matadata对象
print(mapper_register.metadata)
# # 输出：MetaData()

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
5.2： ORM中的模型类
  • 模型类的定义：
from sqlalchemy.orm import registry
from sqlalchemy import create_engine, Column, Integer, VARCHAR, String
import settings

engine = create_engine(settings.DB_URI, echo=True)
# 实例化注册表
mapper_register = registry()
# 得到基类
Base = mapper_register.generate_base()


# 模型类继承于基类
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    age = Column(Integer)

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

  • 解释这个类：

    • 1： Base类中实现了__init__方法，并且会将我们定义的id,name,age 绑定到self中。因此我们可以通过：user = User(id = 1, name = ‘liangdada’， age=23)，这种方式实例化这个对象。
    • 2：Base内部实现了__repr__方法，因此打印模型类对象会输出对象的所有信息。

  • 根据模型类，操作表：

    from sqlalchemy.orm import registry
from sqlalchemy import create_engine, Column, Integer, VARCHAR, String
import settings

engine = create_engine(settings.DB_URI, echo=True)
# 实例化注册表
mapper_register = registry()
# 得到基类
Base = mapper_register.generate_base()


# 模型类继承于基类
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    age = Column(Integer)


# 创建表
mapper_register.metadata.create_all(engine)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21

  • 表反射：

    • 我们清楚了，创建模型类，到根据模型类写入数据库的过程，那么反过来，数据库的一个表，怎么对应我们程序的一个表对象呢？

      some_table = Table("users", metadata, autoload_with=engine)
# 1: 创建表对象，指定表名，指定管理这个表的集合，以及引擎就可以了。
# 2： 实例化这个表对象的时候，会去表集合metadata中查询这个表名，如果存在就去数据库查询这个表的信息，就反射给了这个实例化的表对象。

      • 1
      • 2
      • 3
声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号