SQLAlchemy介绍

SQLAlchemy是一个基于Python的ORM框架。该框架是建立在DB-API之上,使用关系对象映射进行数据库操作。

简而言之就是,将类和对象转换成SQL,然后使用数据API执行SQL并获取执行结果。

补充:什么是DB-API ? 是Python的数据库接口规范。

在没有DB-API之前,各数据库之间的应用接口非常混乱,实现各不相同,

项目需要更换数据库的时候,需要做大量的修改,非常不方便,DB-API就是为了解决这样的问题。

pip install sqlalchemy

组成部分:

  -- engine,框架的引擎

  -- connection pooling  数据库连接池

  -- Dialect  选择链接数据库的DB-API种类(实际选择哪个模块链接数据库)

  -- Schema/Types  架构和类型

  -- SQL Expression Language   SQL表达式语言

连接数据库

SQLAlchemy 本身无法操作数据库,其必须依赖遵循DB-API规范的三方模块,

Dialect 用于和数据API进行交互,根据配置的不同调用不同数据库API,从而实现数据库的操作。

不同的数据库API

# MySQL-PYthon
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>

#pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]

# MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>

# cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

# 更多
# http://docs.sqlalchemy.org/en/latest/dialects/index.html

连接数据库

from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接数pool_size=5,  # 连接池大小pool_timeout=30,  # 连接池中没有线程最多等待时间,否则报错pool_recycle=-1,  # 多久之后对连接池中的连接进行回收(重置)-1不回收
)

执行原生SQL

raw_connection

from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接数pool_size=5,  # 连接池大小pool_timeout=30,  # 连接池中没有线程最多等待时间,否则报错pool_recycle=-1,  # 多久之后对连接池中的连接进行回收(重置)-1不回收
)def test():conn = engine.raw_connection()cursor = conn.cursor()cursor.execute("select * from Course")result = cursor.fetchall()print(result)cursor.close()conn.close()if __name__ == '__main__':test()
# ((1, '生物', 1), (2, '体育', 2), (3, '物理', 1))

contextual_connect

# by gaoxinfrom sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",max_overflow=0,pool_size=5,
)def test():conn = engine.contextual_connect()with conn:cur = conn.execute("select * from Course")result = cur.fetchall()print(result)if __name__ == '__main__':test()
# [(1, '生物', 1), (2, '体育', 2), (3, '物理', 1)]

engine.execute

# by gaoxinfrom sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",max_overflow=0,pool_size=5,
)def test():cur = engine.execute("select * from Course")result = cur.fetchall()print(result)cur.close()if __name__ == '__main__':test()
# [(1, '生物', 1), (2, '体育', 2), (3, '物理', 1)]

ORM

一、创建表

单表的创建

# by gaoxinfrom sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import Index, UniqueConstraint
import datetimeENGINE = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",)Base = declarative_base()class UserInfo(Base):__tablename__ = "user_info"id = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=False)email = Column(String(32), unique=True)create_time = Column(DateTime, default=datetime.datetime.now)__table_args__ = (UniqueConstraint("id", "name", name="uni_id_name"),Index("name", "email"))def create_db():Base.metadata.create_all(ENGINE)def drop_db():Base.metadata.drop_all(ENGINE)if __name__ == '__main__':create_db()

一对多的创建

# by gaoxinfrom sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import Index, UniqueConstraint, ForeignKey
from sqlalchemy.orm import relationship
import datetimeENGINE = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",)Base = declarative_base()# ======一对多示例=======
class UserInfo(Base):__tablename__ = "user_info"id = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=False)email = Column(String(32), unique=True)create_time = Column(DateTime, default=datetime.datetime.now)# FK字段的建立hobby_id = Column(Integer, ForeignKey("hobby.id"))# 不生成表结构 方便查询使用hobby = relationship("Hobby", backref="user")__table_args__ = (UniqueConstraint("id", "name", name="uni_id_name"),Index("name", "email"))class Hobby(Base):__tablename__ = "hobby"id = Column(Integer, primary_key=True)title = Column(String(32), default="码代码")def create_db():Base.metadata.create_all(ENGINE)def drop_db():Base.metadata.drop_all(ENGINE)if __name__ == '__main__':create_db()# drop_db()

多对多的创建

# by gaoxinfrom sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import Index, UniqueConstraint, ForeignKey
from sqlalchemy.orm import relationship
import datetimeENGINE = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",)Base = declarative_base()# ======多对多示例=======
class Book(Base):__tablename__ = "book"id = Column(Integer, primary_key=True)title = Column(String(32))# 不生成表字段 仅用于查询方便tags = relationship("Tag", secondary="book2tag", backref="books")class Tag(Base):__tablename__ = "tag"id = Column(Integer, primary_key=True)title = Column(String(32))class Book2Tag(Base):__tablename__ = "book2tag"id = Column(Integer, primary_key=True)book_id = Column(Integer, ForeignKey("book.id"))tag_id = Column(Integer, ForeignKey("tag.id"))def create_db():Base.metadata.create_all(ENGINE)def drop_db():Base.metadata.drop_all(ENGINE)if __name__ == '__main__':create_db()# drop_db()

二、对数据库表的操作(增删改查)

session

# by gaoxinfrom sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models_demo import TagENGINE = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",)Session = sessionmaker(bind=ENGINE)# 每次执行数据库操作的时候,都需要创建一个session
session = Session()# =======执行ORM操作==========
tag_obj = Tag(title="SQLAlchemy")
# 添加
session.add(tag_obj)
# 提交
session.commit()
# 关闭session
session.close()

scoped_session

# by gaoxinfrom sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from models_demo import TagENGINE = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",)Session = sessionmaker(bind=ENGINE)# 每次执行数据库操作的时候,都需要创建一个session
session = Session()
# 线程安全,基于本地线程实现每个线程用同一个session
# 特殊的:scoped_session中有原来Session中的以下方法
public_methods = {'__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested','close', 'commit', 'connection', 'delete', 'execute', 'expire','expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind','is_modified', 'bulk_save_objects', 'bulk_insert_mappings','bulk_update_mappings','merge', 'query', 'refresh', 'rollback','scalar'
}session = scoped_session(Session)# =======执行ORM操作==========
tag_obj = Tag(title="SQLAlchemy")
# 添加
session.add(tag_obj)
# 提交
session.commit()
# 关闭session
session.close()

基本的增删改查

# by gaoxinfrom sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from models_demo import Tag, UserInfo
import threadingENGINE = create_engine("mysql+pymysql://root:root1234@127.0.0.1:3306/code_record?charset=utf8",)Session = sessionmaker(bind=ENGINE)# 每次执行数据库操作的时候,都需要创建一个session
session = Session()
session = scoped_session(Session)# ============添加============
# tag_obj = Tag(title="SQLAlchemy")
# # 添加
# session.add(tag_obj)
# session.add_all([
#     Tag(title="Python"),
#     Tag(title="Django"),
# ])
# # 提交
# session.commit()
# # 关闭session
# session.close()# ============基础查询============
# ret1 = session.query(Tag).all()
# ret2 = session.query(Tag).filter(Tag.title == "Python").all()
# ret3 = session.query(Tag).filter_by(title="Python").all()
# ret4 = session.query(Tag).filter_by(title="Python").first()
# print(ret1, ret2, ret3, ret4)# ============删除===========
# session.query(Tag).filter_by(id=1).delete()
# session.commit()# ===========修改===========
session.query(Tag).filter_by(id=22).update({Tag.title: "LOL"})
session.query(Tag).filter_by(id=23).update({"title": "王者毒药"})
session.query(Tag).filter_by(id=24).update({"title": Tag.title + "~"}, synchronize_session=False)
# synchronize_session="evaluate" 默认值进行数字加减
session.commit()

常用操作

# 条件查询
ret1 = session.query(Tag).filter_by(id=22).first()
ret2 = session.query(Tag).filter(Tag.id > 1, Tag.title == "LOL").all()
ret3 = session.query(Tag).filter(Tag.id.between(22, 24)).all()
ret4 = session.query(Tag).filter(~Tag.id.in_([22, 24])).first()
from sqlalchemy import and_, or_
ret5 = session.query(Tag).filter(and_(Tag.id > 1, Tag.title == "LOL")).first()
ret6 = session.query(Tag).filter(or_(Tag.id > 1, Tag.title == "LOL")).first()
ret7 = session.query(Tag).filter(or_(Tag.id>1,and_(Tag.id>3, Tag.title=="LOL")
)).all()
# 通配符
ret8 = session.query(Tag).filter(Tag.title.like("L%")).all()
ret9 = session.query(Tag).filter(~Tag.title.like("L%")).all()
# 限制
ret10 = session.query(Tag).filter(~Tag.title.like("L%")).all()[1:2]
# 排序
ret11 = session.query(Tag).order_by(Tag.id.desc()).all()  # 倒序
ret12 = session.query(Tag).order_by(Tag.id.asc()).all()  # 正序
# 分组
ret13 = session.query(Tag.test).group_by(Tag.test).all()
# 聚合函数
from sqlalchemy.sql import func
ret14 = session.query(func.max(Tag.id),func.sum(Tag.test),func.min(Tag.id)
).group_by(Tag.title).having(func.max(Tag.id > 22)).all()
# 连表
ret15 = session.query(UserInfo, Hobby).filter(UserInfo.hobby_id == Hobby.id).all()
# print(ret15) 得到一个列表套元组 元组里是两个对象
ret16 = session.query(UserInfo).join(Hobby).all()
# print(ret16) 得到列表里面是前一个对象
# 相当于inner join
# for i in ret16:
#     # print(i[0].name, i[1].title)
#     print(i.hobby.title)
ret17 = session.query(Hobby).join(UserInfo, isouter=True).all()
ret17_1 = session.query(UserInfo).join(Hobby, isouter=True).all()
ret18 = session.query(Hobby).outerjoin(UserInfo).all()
ret18_1 = session.query(UserInfo).outerjoin(Hobby).all()
# 相当于left join
print(ret17)
print(ret17_1)
print(ret18)
print(ret18_1)

基于relationship的FK

# 基于relationship的FK
# 添加
user_obj = UserInfo(name="提莫", hobby=Hobby(title="种蘑菇"))
session.add(user_obj)hobby = Hobby(title="弹奏一曲")
hobby.user = [UserInfo(name="琴女"), UserInfo(name="妹纸")]
session.add(hobby)
session.commit()# 基于relationship的正向查询
user_obj_1 = session.query(UserInfo).first()
print(user_obj_1.name)
print(user_obj_1.hobby.title)# 基于relationship的反向查询
hb = session.query(Hobby).first()
print(hb.title)
for i in hb.user:print(i.name)session.close()

基于relationship的M2M

# 添加
book_obj = Book(title="Python源码剖析")
tag_obj = Tag(title="Python")
b2t = Book2Tag(book_id=book_obj.id, tag_id=tag_obj.id)
session.add_all([book_obj,tag_obj,b2t,
])
session.commit()#  上面有坑哦~~~~
book = Book(title="测试")
book.tags = [Tag(title="测试标签1"), Tag(title="测试标签2")]
session.add(book)
session.commit()tag = Tag(title="LOL")
tag.books = [Book(title="大龙刷新时间"), Book(title="小龙刷新时间")]
session.add(tag)
session.commit()# 基于relationship的正向查询
book_obj = session.query(Book).filter_by(id=4).first()
print(book_obj.title)
print(book_obj.tags)
# 基于relationship的反向查询
tag_obj = session.query(Tag).first()
print(tag_obj.title)
print(tag_obj.books)

转载于:https://www.cnblogs.com/cainingning/p/9957442.html

SQLAlchemy介绍相关推荐

  1. python学习之flask sqlalchemy介绍和使用 表操作增删查改

    flask sqlalchemy介绍和使用 表操作增删查改 内容详细 1 sqlalchemy介绍和快速使用 1.1 介绍 # SQLAlchemy是一个基于Python实现的ORM框架# djang ...

  2. Python SQLAlchemy介绍

    Python SQLAlchemy教程 一. 介绍 关于ORM 二. 使用 概念和数据类型 概念 常见数据类型 使用步骤 创建数据库表 1.安装 2. 创建连接 3. 创建数据库表类(模型) 4. 生 ...

  3. Python SQLAlchemy

    SQLAlchemy介绍 SQLAlchemy是一个基于Python的ORM框架.该框架是建立在DB-API之上,使用关系对象映射进行数据库操作. 简而言之就是,将类和对象转换成SQL,然后使用数据A ...

  4. python3之SQLAlchemy

    1.SQLAlchemy介绍 SQLAlchemy是Python SQL工具包和对象关系映射器,为应用程序开发人员提供了SQL的全部功能和灵活性. 它提供了一整套众所周知的企业级持久性模式,专为高效和 ...

  5. SQLAlchemy基本使用(Flask中)

    SQLAlchemy介绍 SQLAlchemy是一个基于Python实现的ORM框架. 该框架建立在 DB API之上,使用关系对象映射进行数据库操作,简言之便是:将类和对象转换成SQL,然后使用数据 ...

  6. SQLAlchemy简单入门

    SQlAlchemy简单使用 sqlalchemy介绍 SQLAlchemy的是Python的SQL工具包和对象关系映射器,让应用程序开发人员可以使用上SQL的强大功能和灵活性. 它提供了一套完整的企 ...

  7. 8Flask-----------Flask框架------------安装使用、基本介绍

    Flask 一.Flask 简介 (一) 环境搭建 (二)运行项目 二.项⽬内容介绍 (一)app.route路由URL设定 1)传参 2)请求方式 (二)def视图函数 1)构造URL(url_fo ...

  8. 软件测试 | 测试开发 | Aiomysql 与 Sqlalchemy 的使用

    背景 之前一直使用tornado做项目,数据库一直使用mongo与redis居多,凭借其优异的异步特性工作的也很稳定高效,最近的项目需要使用 mysql ,由于之前在使用mongo与redis时所使用 ...

  9. database - sqlalchemy

    1.SQLAlchemy介绍 SQLAlchemy是Python中最有名的ORM工具,ORM全称Object Relational Mapping(对象关系映射) ORM操作是所有完整软件中后端处理最 ...

  10. python3入门到精通pdf-Python3入门到精通实战特训

    原标题:Python3入门到精通实战特训 随着时间的推移,python2.7版本很快就要停止官方更新了.如果你还没来及正式考虑版本迁移问题,或者你是一个python新手.那么就很有必要一试本课程,一方 ...

最新文章

  1. Jupyter 官方神器:可视化 Debug 工具!
  2. 某央企程序员炫耀:央企研发岗才是性价比天花板!955,不卷不裁,6险2金,500元单身公寓!...
  3. error LNK2019: unresolved external symbol _WinMain@16 referenced in function ___tmainCRTStartup
  4. OSError: mysql_config not found
  5. 提取文件出错_提取中文、英文和数字,其实很简单
  6. hashcat离线碰撞md5
  7. 名词解释(容器、并发,插件,脚本)及程序对象的创建和注释文档
  8. 局部变量和参数传递的问题
  9. 系统架构师学习笔记-操作系统(二)
  10. 同步互斥阻塞 (2)
  11. SPSS22.0简体中文破解版(32位/64位)使用方法
  12. 项目管理--maven浅析《四》之私服(Nexus)
  13. 中国最新7座超大城市、14座特大城市完整榜单
  14. 有关Cidaemon.exe服务占用大量CPU资源的解决办法
  15. Kanzi Shader入门
  16. 苹果手机录音怎么批量导出?才知道原来可以这样操作
  17. Win11机械硬盘磁盘碎片整理方法
  18. 统计分布——常用且重要的三种分布
  19. AJAX简单思维导图
  20. MATLAB 数据频数统计

热门文章

  1. 邮件服务器搬家,企业邮箱怎么“搬家”
  2. 协作,才能更好的中断线程
  3. springboot获取Spring容器中的bean(ApplicationContextAware接口的应用)避免过多的或不用if..else,switch
  4. ubuntu 安装mysql 5.5.28 编译安装 innodb 配置
  5. 阶段5 3.微服务项目【学成在线】_day04 页面静态化_13-页面静态化-数据模型-轮播图DataUrl接口...
  6. 阶段5 3.微服务项目【学成在线】_day02 CMS前端开发_04-vuejs研究-vuejs基础-v-model指令...
  7. 小D课堂 - 零基础入门SpringBoot2.X到实战_第4节 Springboot2.0单元测试进阶实战和自定义异常处理_21、SpringBoot2.x配置全局异常返回自定义页面...
  8. 阶段3 1.Mybatis_09.Mybatis的多表操作_2 完成account表的建立及实现单表查询
  9. 在上传前探测磁盘是否挂载正常
  10. LNMP下Redis介绍以及安装(Linux)