一、介绍
基于Python语言,是Python的一个ORM框架,ORM将对象的各个方法翻译成SQL语句,通过DB-API执行。
DB-API : 是Python的数据库接口规范
二、下载
pip install sqlalchemy
三、单表的创建
# 导入SQLAlchemy
from sqlalchemy import Column,String,Integer
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
# 创建对象的基类
Base = declarative_base()
# 初始化数据库连接
engine = create_engine(
"mysql+pymysql://root:@127.0.0.1:3306/mysqlalchemydb?charset=utf8",
max_overflow = 0, # 超过连接池大小外最多创建的连接数
pool_size = 5, # 连接池大小
pool_timeout = 30, # 连接池中没有线程最多等待时间,否则报错
pool_recycle = -1 # 多久之后对连接池中的连接进行回收(重置)-1不回收
)
class Book(Base):
# 表的名字
__tablename__ = 'book'
# 表的结构
id = Column(Integer,primary_key=True)
title = Column(String(32),nullable=True)
if __name__ == '__main__':
# 创建表
Base.metadata.create_all(engine)
# 删除表
Base.metadata.drop_all(engine)
四、一对多的创建和数据添加
# 导入SQLAlchemy
from sqlalchemy import Column,String,Integer,ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
# 创建对象的基类
Base = declarative_base()
# 初始化数据库连接
engine = create_engine(
"mysql+pymysql://root:@127.0.0.1:3306/mysqlalchemydb?charset=utf8",
max_overflow = 0, # 超过连接池大小外最多创建的连接数
pool_size = 5, # 连接池大小
pool_timeout = 30, # 连接池中没有线程最多等待时间,否则报错
pool_recycle = -1 # 多久之后对连接池中的连接进行回收(重置)-1不回收
)
class Book(Base):
# 表的名字
__tablename__ = 'book'
# 表的结构
id = Column(Integer,primary_key=True)
title = Column(String(32),nullable=True)
publisher_id = Column(Integer,ForeignKey("publisher.id"))
# relationship 不生成字段建立关系,用于操作
# backref 反向查询的字段
publisher = relationship("Publisher",backref="books")
class Publisher(Base):
# 表的名字
__tablename__ = 'publisher'
# 表的结构
id = Column(Integer,primary_key=True)
title = Column(String(32),nullable=True)
from sqlalchemy.orm import sessionmaker,scoped_session
# 绑定数据库连接,生成管理器对象,但是线程不安全
Session = sessionmaker(bind=engine)
# 重新创造对象保证线程安全
session = scoped_session(Session)
# # 创建表
Base.metadata.create_all(engine)
# # 删除表
# Base.metadata.drop_all(engine)
# 生成对象,并用外键id创建数据
publisher_obj = Publisher(title="深圳出版社")
book_obj = Book(title="数学之美",publisher_id=1) # 存放外键的id
session.add(publisher_obj)
session.commit()
session.close()
# # 生成对象,并用外键对象创建数据
publisher_obj = Publisher(title="上海出版社")
book_obj = Book(title="语文之美",publisher=publisher_obj) # 存放外键的id
session.add(publisher_obj)
session.commit()
session.close()
五、多对多关系表的创建
# 导入SQLAlchemy
from sqlalchemy import Column,String,Integer,ForeignKey,UniqueConstraint
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
# 创建对象的基类
Base = declarative_base()
# 初始化数据库连接
engine = create_engine(
"mysql+pymysql://root:@127.0.0.1:3306/mysqlalchemydb?charset=utf8",
max_overflow = 0, # 超过连接池大小外最多创建的连接数
pool_size = 5, # 连接池大小
pool_timeout = 30, # 连接池中没有线程最多等待时间,否则报错
pool_recycle = -1 # 多久之后对连接池中的连接进行回收(重置)-1不回收
)
class Book(Base):
# 表的名字
__tablename__ = 'book'
# 表的结构
id = Column(Integer,primary_key=True)
title = Column(String(32),nullable=True)
# relationship 不生成字段建立关系,用于操作
# backref 反向查询的字段
# secondary 关联第三张表
tags = relationship("Tag",backref="books",secondary="book2tag")
# 只有__repr__,面向开发人员
def __repr__(self):
return self.title
class Tag(Base):
__tablename__ = "tag"
id = Column(Integer, primary_key=True)
title = Column(String(32), nullable=True)
# 只有__repr__,面向开发人员
def __repr__(self):
return self.title
__table_args__ = (
# 联合唯一
UniqueConstraint("id","title",name="uni_id_title"),
)
# 多对多关系关联的表
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"))
from sqlalchemy.orm import sessionmaker,scoped_session
# 绑定数据库连接,生成管理器对象,但是线程不安全
Session = sessionmaker(bind=engine)
# 重新创造对象保证线程安全
session = scoped_session(Session)
# # 创建表
Base.metadata.create_all(engine)
# # 删除表
# Base.metadata.drop_all(engine)
六、基本操作
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
# 导入SQLAlchemy
from sqlalchemy import Column,String,Integer,ForeignKey,UniqueConstraint
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
# 创建对象的基类
Base = declarative_base()
# 初始化数据库连接
engine = create_engine(
"mysql+pymysql://root:@127.0.0.1:3306/mysqlalchemydb?charset=utf8",
max_overflow = 0, # 超过连接池大小外最多创建的连接数
pool_size = 5, # 连接池大小
pool_timeout = 30, # 连接池中没有线程最多等待时间,否则报错
pool_recycle = -1 # 多久之后对连接池中的连接进行回收(重置)-1不回收
)
class Book(Base):
# 表的名字
__tablename__ = 'book'
# 表的结构
id = Column(Integer,primary_key=True)
title = Column(String(32),nullable=True)
# relationship 不生成字段建立关系,用于操作
# backref 反向查询的字段
# secondary 关联第三张表
tags = relationship("Tag",backref="books",secondary="book2tag")
# 只有__repr__,面向开发人员
def __repr__(self):
return self.title
class Tag(Base):
__tablename__ = "tag"
id = Column(Integer, primary_key=True)
title = Column(String(32), nullable=True)
# 只有__repr__,面向开发人员
def __repr__(self):
return self.title
__table_args__ = (
# 联合唯一
UniqueConstraint("id","title",name="uni_id_title"),
)
# 多对多关系关联的表
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"))
from sqlalchemy.orm import sessionmaker,scoped_session
# 绑定数据库连接,生成管理器对象,但是线程不安全
Session = sessionmaker(bind=engine)
# 重新创造对象保证线程安全
session = scoped_session(Session)
增:
# 添加
tag_obj = Tag(title="基础")
tag_obj2 = Tag(title="进阶")
tag_obj3 = Tag(title="高级")
session.add(tag_obj) # 添加单个
session.add_all([tag_obj2,tag_obj3]) # 批量添加
session.commit() # 提交
session.close() # 关闭session
查:
# 查询
ret1 = session.query(Tag).all()
print(ret1) # [基础, 进阶, 高级]
ret2 = session.query(Tag).filter(Tag.id == 2).first()
ret3 = session.query(Tag).filter_by(id=3).first()
print(ret2) # 进阶
print(ret3) # 高级
改:
# 修改
session.query(Tag).filter_by(id=1).update({"title":"python"})
session.query(Tag).filter_by(id=2).update({Tag.title:"Linux"})
session.commit() # 提交
session.close() # 关闭
删:
# 删除
session.query(Tag).filter_by(id=3).delete()
session.commit()
session.close()