您现在的位置是：网站首页> 编程资料编程资料

Python ORM数据库框架Sqlalchemy的使用教程详解_python_

2023-05-25 226人已围观

简介 Python ORM数据库框架Sqlalchemy的使用教程详解_python_

对象关系映射（Object Relational Mapping，简称ORM）模式是一种为了解决面向对象与关系数据库存在的互不匹配的现象的技术。面向对象的开发方法是当今企业级应用开发环境中的主流开发方法，关系数据库是企业级应用环境中永久存放数据的主流数据存储系统。对象和关系数据是业务实体的两种表现形式，业务实体在内存中表现为对象，在数据库中表现为关系数据。内存中的对象之间存在关联和继承关系，而在数据库中，关系数据无法直接表达多对多关联和继承关系。因此，对象-关系映射(ORM)系统一般以中间件的形式存在，主要实现程序对象到关系数据库数据的映射。

学过java的hibernate框架的那么这个很好上手,非常简单 ,他有两种模式一种纯orm另一种模式是支持原生sql这两种可以混合使用

优点：

简洁易读：将数据表抽象为对象（数据模型），更直观易读
可移植：封装了多种数据库引擎，面对多个数据库，操作基本一致，代码易维护
更安全：有效避免SQL注入

缺点: 虽然性能稍稍不及原生SQL，但是操作数据库真的很方便！

官网: https://www.sqlalchemy.org/

概念和数据类型

概念

常见数据类型

安装

pip install SQLAlchemy pip install mysqlclient # 安装自己的数据库客户端(可以是mysql 可以是oracle)

连接

from sqlalchemy import create_engine engine = create_engine("mysql://user:password@hostname/dbname?charset=uft8", echo=True, pool_size=8, pool_recycle=60*30 )

创建好了Engine的同时，Pool和Dialect也已经创建好了，但是此时并没有真正与数据库连接，等到执行具体的语句.connect()等时才会连接到数据库。

echo: 当设置为True时会将orm语句转化为sql语句打印，一般debug的时候可用
pool_size: 连接池的大小，默认为5个，设置为0时表示连接无限制
pool_recycle: 设置时间以限制数据库多久没连接自动断开

创建数据库表类（模型）

ORM的重要特点就是操作类来操作数据库，现在我们来创建一个类，以常见的用户表举例：

from sqlalchemy import Column, Integer, String from src.database.SQLalchemyFast import SQLalchemyFast class UserDB(SQLalchemyFast.Base): __tablename__ = "User" # __tablename__ 声明表名 # primary_key为主键,autoincrement为自增, doc 为注释但是不会在数据库中生成 id = Column(Integer, primary_key=True,autoincrement=True,doc="主键") name = Column(String(64), unique=True, doc="用户名") # unique=True 为唯一约束会在数据库中生成索引 email = Column(String(64), doc="邮箱") def __init__(self, name=None, email=None): self.name = name self.email = email def __str__(self): return "UserDB(id=%s,name=%s,email=%s)" % (self.id, self.name, self.email)

上面的SQLalchemyFast.Base是我自己封装的Base ,用于统一管理所有模型类,可以将Python类和数据库表关联映射起来。数据库表模型类通过__tablename__和表关联起来，Column表示数据表的列

生成数据库表

Base = declarative_base() Base.metadata.create_all(engine)

会自动创建表，如果存在则忽略，执行以上代码，就会发现在db中创建了users表。前提必须有模型类继承了Base

会话

会话就和打电话一样,打一次电话就是一个会话,就相当于和数据库交互一次就是一个会话,一个会话可以运行多个或单个语句,会话结束必须关闭

sqlalchemy中使用session用于创建程序和数据库之间的会话，所有对象的载入和保存都需要通过session对象。

通过sessionmaker调用创建一个工厂，并关联Engine以确保每个session都可以使用该Engine连接资源：

from sqlalchemy.orm import sessionmaker # 创建session DbSession = sessionmaker(bind=engine) session = DbSession()

session的常见操作方法包括：

flush：预提交，提交到数据库文件，还未写入数据库文件中 (没事用)
commit：提交了一个事务
rollback：回滚
close：关闭

增删改查

增

add_user = UserDB("test", "test123@qq.com") session.add(add_user) session.commit()

session.add()将会把Model加入当前session维护的持久空间(可以从session.dirty看到)中，直到commit时提交到数据库。

查

users = session.query(UserDB).filter(UserDB.id=1).all() for item in users: print(item.name)

改

session.query(Users).filter(UserDB.id=1).update({'name': "Jack"})

删

session.query(UserDB).filter(UserDB.name == "test").delete() session.commit()

执行裸sql

session.execute(text(sql), params) session.commit()

sql: select * from User where id = :id and name = :name

params: {"id":1,"name":"张三"}`

参数名必须和sql语句中的参数名一致

with关闭会话

 DbSession = sessionmaker(bind=engine) with DbSession() as conn: # 代码 conn.commit()

sql建造者模式

需要导入的包

from sqlalchemy import delete, update, text, select, join, desc, asc sql = select(UserDB.id,UserDB.name).select_from(UserDB).\ where(text("id = :id and name = :name")).\ group_by(UserDB.id,UserDB.name).\ having(text("id = :id and name = :name")).\ order_by(desc("id"),asc("name")).\ offset(1).limit(10).\ params(id=1, name="张三") print(sql)

以上sql放入到execute里直接就能跑了

多表联查(只支持内查询和左查询和全查询)

 sql = select(UserDB.id,UserDB.name).select_from(UserDB).\ join(BookDB,UserDB.id == BookDB.id).\ join(alias(BookDB,"b"),text("b.id == b.id"),isouter=True).\ join(alias(BookDB,"e"),text("e.id == e.id"),full=True). \ where(text("id = :id and name = :name")) print(sql)

封装的工具

数据库配置文件database.properties

url=mysql://root:root@106.12.174.220:3306/demo?charset=utf8 echo=True # 是否打印sql语句 pool_size=10 # 连接池大小 pool_recycle=1800 # 连接池回收时间 pool_timeout=30 # 连接池超时时间 isolation_level=READ_COMMITTED # 事务隔离级别

工具

from sqlalchemy import create_engine, delete, update, text, alias from sqlalchemy.future import select from sqlalchemy.orm import declarative_base, sessionmaker from src.file.FileReadAndWrite import FileReadAndWrite from src.log.Log import Log class SQLalchemyFast(object): Base = declarative_base() """ 功能: SQLalchemy工具 """ def __init__(self, dbFile): file = FileReadAndWrite.readPropertiesFile(dbFile) self.engine = create_engine( url=file['url'], echo=bool(file['echo']), pool_size=int(file['pool_size']), pool_recycle=int(file['pool_recycle']), pool_timeout=int(file['pool_timeout']), isolation_level=file['isolation_level'], ) SQLalchemyFast.Base.metadata.create_all(self.engine) # 创建表,如果表存在则不创建(必须对象继承Base) # 创建会话 def createSession(self): Session = sessionmaker(bind=self.engine) return Session() # 添加一条数据 def addData(self, object): with self.createSession() as conn: conn.add(object) conn.commit() # 添加多条数据 def addDataList(self, objectList): with self.createSession() as conn: conn.add_all(objectList) conn.commit() # 删除主键id的数据 def deleteDataById(self, cla, id): with self.createSession() as conn: conn.query(cla).filter(cla.id == id).delete() conn.commit() # 删除指定数据(where是并且的关系,不支持or和其他复杂查询) def deleteDataWhere(self, cla, *where): with self.createSession() as conn: stmt = delete(cla).where(*where) conn.execute(stmt) conn.commit() # 清空表 def truncateTable(self, cla): with self.createSession() as conn: conn.query(cla).delete() conn.commit() # 更新指定主键id的数据 def updateDataById(self, cla, id, data): """ :param cla: 类(表) :param id: 主键id :param data: {'key': "value",...} key为表中的字段名,value为要修改的值 :return: """ with self.createSession() as conn: stmt = update(cla).where(cla.id == id).values(data) result = conn.execute(stmt) conn.commit() return result # 更新指定条件的数据 (where是并且的关系,不支持or和其他复杂查询) def updateDataWhere(self, cla, data, *where): """ :param cla: 类(表) :param data: {'key': "value",...} key为表中的字段名,value为要修改的值 :param where: 过滤条件 :return: """ with self.createSession() as conn: stmt = update(cla).where(*where).values(data) conn.execute(stmt) conn.commit() # 查询全部数据 def queryDataAll(self, cla): with self.createSession() as conn: result = conn.query(cla).all() return result # 查询主键id的数据 def queryDataById(self, cla, id): with self.createSession() as conn: result = conn.query(cla).filter(cla.id == id).first() return result # 查询指定数据,不支持分组查询(因为聚合后的数据无法转换成对象) def queryDataWhere(self, cla,aliasName=None, column=None, where=None, join=None, on=None, left=None, full=None, order="", limit="", offset="", distinct=None, params=None): with self.createSession() as conn: stmt = select(cla) if aliasName: stmt = select(alias(cla,aliasName)) if column: stmt = stmt.with_only_columns(text
                提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                
                
                                            上一篇：详解Python AdaBoost算法的实现_python_
                                                                下一篇：Python+Pillow进行图形处理的示例详解_python_