SQLAlchemy是应用于python上的ORM框架。

1.ORM

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

2.SQLAlchemy安装

#安装sqlalchemy模块 
pip install sqlalchemy  

1.始化DBSession。

# -*- coding: utf-8 -*-
# @Time : 2022/5/5 18:32
# @File : sqlalchemydemo.py
# @Software : PyCharm

from sqlalchemy import create_engine
# 引擎
from sqlalchemy.orm import sessionmaker


# 创建orm的会话池，orm和sql均可以管理对象关系型数据库，需要绑定引擎才可以使用会话，
# 创建连接
engine = create_engine("mysql+pymysql://root:root@127.0.0.1/test",
                       # 需要安装mysql和pymysql的模块，用户名:密码@ip地址/某个数据库
                       #echo=True,         # 打印操作对应的SQL语句
                       pool_size=8,       # 连接个数
                       pool_recycle=60*30 # 不使用时断开
                       )

# 创建session
DbSession = sessionmaker(bind=engine)
# 会话工厂，与引擎绑定。
# 实例化
session = DbSession()
# 关闭会话
session.close()

2.创建books表。

# -*- coding: utf-8 -*-
# @Time : 2022/5/5 18:43
# @File : sqlalchemydemo2.py
# @Software : PyCharm


from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String

engine = create_engine('mysql+pymysql://root:root@127.0.0.1/test')
metadata = MetaData(engine)

student = Table('books', metadata,
                Column('id', Integer, primary_key=True),
                Column('name', String(50)),
                Column('auth', String(50))
                )

metadata.create_all(engine)

3.插入数据

# -*- coding: utf-8 -*-
# @Time : 2022/5/5 18:43
# @File : sqlalchemydemo3.py
# @Software : PyCharm

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('mysql+pymysql://root:root@127.0.0.1/test')
DBsession = sessionmaker(bind=engine)
session = DBsession()

Base = declarative_base()


class Books(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    auth = Column(String(50))


'''指定主键方式
book1 = Books(id=1, name='Thinking in Java', auth="Bob")
book2 = Books(id=2, name='Javascript Core', auth="Alan")
book3 = Books(id=3, name='计算机网络', auth="雷震甲")
'''

book1 = Books(name='Thinking in Java', auth="Bob")
book2 = Books(name='Javascript Core', auth="Alan")
book3 = Books(name='计算机网络', auth="雷震甲")

session.add_all([book1, book2, book3])
session.commit()
session.close()

4.查询所有图书

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('mysql+pymysql://root:root@127.0.0.1/test')
DBsession = sessionmaker(bind=engine)
session = DBsession()

Base = declarative_base()


class Books(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    auth = Column(String(50))

# 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
books = session.query(Books).all()
# 打印类型和对象的name属性:
for b in books:
    print(b.id,',',b.name,',',b.auth)
session.commit()
session.close()

5.一个完整的CRUD案例

import traceback
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import sessionmaker

# 创建orm的会话池，orm和sql均可以管理对象关系型数据库，需要绑定引擎才可以使用会话，
# 创建连接
engine = create_engine("mysql+pymysql://root:root@127.0.0.1/test",
                       # 需要安装mysql和pymysql的模块，用户名:密码@ip地址/某个数据库
                       # echo=True,         # 打印操作对应的SQL语句
                       pool_size=8,  # 连接个数
                       pool_recycle=60 * 30  # 不使用时断开
                       )
DBsession = sessionmaker(bind=engine)
session = DBsession()

Base = declarative_base()


class Users(Base):  # 继承生成的orm基类
    __tablename__ = "users"  # 表名
    uid = Column(Integer, primary_key=True, autoincrement=True)  # 设置主键
    username = Column(String(20))
    password = Column(String(50))

    def __str__(self):
        return "用户编号：%d,用户名：%s,密码：%s" % (self.uid, self.username, self.password)


# 执行DDL语句创建用户表
def execute_ddl():
    try:
        Base.metadata.create_all(engine)  # 创建表结构
    except Exception as e:
        '''如果发生错误则回滚并输出提示语句'''
        msg = traceback.format_exc()
        print(msg)
        print("创建用户表失败")


def execute_insert(user):
    try:
        session.add(user)
        session.commit()
        print('新增用户资料成功')
    except Exception as e:
        '''如果发生错误则回滚并输出提示语句'''
        session.rollback()
        msg = traceback.format_exc()
        print(msg)
        print("插入用户资料失败")
    finally:
        session.close()


# 执行批量新增用户
def execute_batch(user_list):
    try:
        session.add_all(user_list)
        session.commit()
        print('新增用户集合资料成功')
    except Exception as e:
        '''如果发生错误则回滚并输出提示语句'''
        msg = traceback.format_exc()
        session.rollback()
        print(msg)
        print("插入用户集合资料失败")
    finally:
        session.close()


# 执行查询语句(查询单个用户)
def execute_query(uid):
    try:
        user = session.query(Users).where(Users.uid == uid).one()
        return user
    except Exception as e:
        '''如果发生错误则回滚并输出提示语句'''
        session.rollback()
        msg = traceback.format_exc()
        print(msg)
        print("查询资料失败")
    finally:
        session.close()


# 执行查询语句(查询所有用户)
def execute_queryAll():
    try:
        user_list = session.query(Users).all()
        return user_list
    except Exception as e:
        '''如果发生错误则回滚并输出提示语句'''
        session.rollback()
        msg = traceback.format_exc()
        print(msg)
        print("查询资料失败")
    finally:
        session.close()


# 更新用户资料
def execute_update(user):
    try:
        session.add(user)
        session.commit()
        print('更新用户资料成功!')
    except Exception as e:
        '''如果发生错误则回滚并输出提示语句'''
        session.rollback()
        msg = traceback.format_exc()
        print(msg)
        print("更新用户资料失败")
    finally:
        session.close()

#删除用户资料
def execute_remove(uid):
    try:
        session.delete(uid)
        session.commit()
        print('删除用户资料成功!')
    except Exception as e:
        '''如果发生错误则回滚并输出提示语句'''
        session.rollback()
        msg = traceback.format_exc()
        print(msg)
        print("删除用户资料失败")
    finally:
        session.close()


if __name__ == '__main__':
    # user_list = []
    # execute_ddl()

    # user = Users(uid=100, username='张飞', password='123456')
    # user = Users(username='张飞', password='123456')
    '''
    user_list.append(Users(username='刘备', password='123456'))
    user_list.append(Users(username='关羽', password='654321'))
    user_list.append(Users(username='赵云', password='888888'))
    user_list.append(Users(username='马超', password='2222222'))
    '''
    # execute_insert(user)    # execute_batch(user_list)

    # user = execute_query(5)
    # print(user)
    '''
    user_list = execute_queryAll()
    for u in user_list:
        print(u)
    '''

    '''
    user = execute_query(5)
    print(user)
    user.username = "马化腾"
    user.password = '999999'
    execute_update(user)
    '''
    user = execute_query(1)
    execute_remove(user)

小结: python中最有名的ORM架构就是SQLAlchemy，所谓ORM，就是Object Relationship Mapping对象模型与数据库表的映射。