SQLAlchemy快速入门

安装依赖

pip install sqlalchemy
pip install pymysql

创建数据库和表

# 创建数据库
drop database if exists sqlalchemy_demo;
create database sqlalchemy_demo character set utf8mb4;
use sqlalchemy_demo;# 创建表
drop table if exists user;
create table user
(id   int primary key auto_increment,name varchar(36)
) engine = innodbcharacter set utf8mb4;

新增用户

新增用户的核心代码是session.add(new_user)，需要注意的是，在执行了增加以后，一定要紧接着执行session.commit()提交事务，否则新增将不会生效。

示例代码：

from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, declarative_base# 创建对象的基类:
Base = declarative_base()# 定义User对象:
class User(Base):# 表的名字:__tablename__ = 'user'# 表的结构:id = Column(String(20), primary_key=True)name = Column(String(20))# 创建引擎
engine = create_engine("mysql+pymysql://root:zhangdapeng520@localhost:3306/sqlalchemy_demo?charset=utf8mb4")# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)# 创建session对象:
session = DBSession()# 创建新User对象:
new_user = User(id='1', name='张三')# 添加到session:
session.add(new_user)# 提交即保存到数据库:
session.commit()# 关闭session:
session.close()

根据ID查询用户

查询单个用户的核心代码是session.query(User).filter(User.id == '1').one()，通过这种方式，我们能够查询出指定ID的用户。

示例代码：

from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, declarative_base# 创建对象的基类:
Base = declarative_base()# 定义User对象:
class User(Base):# 表的名字:__tablename__ = 'user'# 表的结构:id = Column(String(20), primary_key=True)name = Column(String(20))# 创建引擎
engine = create_engine("mysql+pymysql://root:zhangdapeng520@localhost:3306/sqlalchemy_demo?charset=utf8mb4")# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)# 创建session对象:
session = DBSession()# 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
user = session.query(User).filter(User.id == '1').one()# 打印类型和对象的name属性:
print('type:', type(user))
print('name:', user.name)# 关闭Session:
session.close()

查询所有用户

查询所有用户的核心代码是session.query(User).all()，通过这种方式，我们能够得到一个用户数组，这个数组中包含所有的用户信息。

示例代码：

from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, declarative_base# 创建对象的基类:
Base = declarative_base()# 定义User对象:
class User(Base):# 表的名字:__tablename__ = 'user'# 表的结构:id = Column(String(20), primary_key=True)name = Column(String(20))# 创建引擎
engine = create_engine("mysql+pymysql://root:zhangdapeng520@localhost:3306/sqlalchemy_demo?charset=utf8mb4")# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)# 创建session对象:
session = DBSession()# 查询所有用户
users = session.query(User).all()
print(users)
for u in users:print(u.id, u.name)# 关闭Session:
session.close()

根据id删除用户

根据id删除用户的核心代码是session.query(User).filter(User.id == "1").delete()，需要注意的是，一定要在执行删除以后，继续执行session.commit()提交事务，否则删除操作将不会生效。

示例代码：

from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, declarative_base# 创建对象的基类:
Base = declarative_base()# 定义User对象:
class User(Base):# 表的名字:__tablename__ = 'user'# 表的结构:id = Column(String(20), primary_key=True)name = Column(String(20))# 创建引擎
engine = create_engine("mysql+pymysql://root:zhangdapeng520@localhost:3306/sqlalchemy_demo?charset=utf8mb4")# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)# 创建session对象:
session = DBSession()# 根据ID删除用户
session.query(User).filter(User.id == "1").delete()
session.commit()  # 一定要记得commit，否则不生效# 关闭Session:
session.close()

根据ID更新用户

使用with DBSession() as s格式的with代码块，with会自动帮我们调用close关闭方法，这样我们就不需要关系session的关闭了，能够节省一些代码和逻辑。不过commit方法还是需要我们自己手动进行调用。

更新用户的核心代码是s.query(User).filter(User.id == "1").update({"name": "张三333"})，需要注意的是，update方法跟的是一个字典，在字典中指定需要更新的键值对。

示例代码：

from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, declarative_base# 创建对象的基类:
Base = declarative_base()# 定义User对象:
class User(Base):# 表的名字:__tablename__ = 'user'# 表的结构:id = Column(String(20), primary_key=True)name = Column(String(20))# 创建引擎
engine = create_engine("mysql+pymysql://root:zhangdapeng520@localhost:3306/sqlalchemy_demo?charset=utf8mb4")# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)# 先增加用户
# 使用with的好处是，with代码块会自动帮我们调用close方法
# 注意：commit方法需要我们自己手动调用
with DBSession() as s:s.add(User(id="1", name="张三"))s.commit()# 根据ID更新用户
with DBSession() as s:s.query(User).filter(User.id == "1").update({"name": "张三333"})s.commit()# 根据ID查询用户
with DBSession() as s:user = s.query(User).filter(User.id == "1").one()print(user.id, user.name)