安装

# 安装 SQLAlchemy
pip install sqlalchemy
# 如果用 MySQL，额外安装驱动
pip install pymysql

连接数据库

from sqlalchemy import create_engine

engine = create_engine('mysql://user:pwd@localhost/testdb', echo=True)
connection = engine.connect()

使用 mysql 数据库进行连接需要先安装 mysqlclient

SQLAlchemy 连接 MySQL 时，支持的驱动优先级 / 格式如下：

sql 语句查询

import sqlalchemy

engine = sqlalchemy.create_engine(...)
conn = engine.connect()

query = sqlalchemy.text("SELECT * FROM students")
result_set = conn.execute(query)

for row in result_set:
    print(row)

# 关闭连接
conn.close()

# 销毁引擎
engine.dispose()

创建表

from sqlalchemy

engine = create_engine(..., echo=True)
meta = sqlalchemy.MetaData()

person = sqlalchemy.Table(
    'person', meta,
    sqlalchemy.Column('id', sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column('name', sqlalchemy.String(128), unique=True, nullable=False), # 唯一，且不能为空
    sqlalchemy.Column('birthday', sqlalchemy.Date, nullable=False)
)
meta.create_all(engine)

插入数据

插入单条数据

person_insert = person.insert()
insert_tom = person_insert.values(name="Tom", birthday="2000-10-10")

with engine.connect() as conn:
    result = conn.execute(insert_tom)
    print(result.inserted_primary_key)
    conn.commit()

批量插入

person_insert = person.insert()
with engine.connect() as conn:
    conn.execute(person_insert, [
        {"name": "Jack", "birthday": "2000-10-10"},
        {"name": "Rose", "birthday": "2000-10-10"}
    ])
    conn.commit()

查询数据

with engine.connect() as conn:
    query = person_table.select() # 通过 sqlalchemy.Table() 创建的 Table
    result_set = conn.execute(query)

    for row in result_set: # 获取的是一个迭代器，每次调用提取一条数据，不会一次性的全部提取出来
        print(row)


# 返回数组
result = result_set.fetchall() # 一次性提取所有数据


# 获取第一条数据，返回元组
row = result_set.fetchone()

条件查询

query = person_table.select().where(person_table.c.birthday > '2000-10-13')
result_set = conn.execute(query)
result = result_set.fetchall()

query = person_table.select()
    .where(person_table.c.birthday > '2000-10-13')
    .where(person_table.c.id < 6)
result_set = conn.execute(query)
result = result_set.fetchall()

from sqlalchemy import and_, or_

query = person_table.select()
    .where(
        or_(
            person_table.c.name == "Tome",
            and_(
                person_table.c.birthday > '2000-10-13',
                person_table.c.id < 6
            )
        )
    )

result_set = conn.execute(query)
result = result_set.fetchall()

更新数据

update_query = person_table.update().values(name="asdf") # 更新所有的 name 字段
conn.execute(update_query)
conn.commit()

update_query = person_table.update.values(name="a").where(person_table.c.id == 6)
conn.execute(update_query)
conn.commit()

删除数据

# 删除所有数据
delete_query = person_table.delete()
conn.execute(delete_query)
conn.commit()

# 删除 id 为 6 的数据
delete_query = person_table.delete().where(person_table.c.id == 6)
conn.execute(delete_query)
conn.commit()

关联表

一对多

department_table = sqlalchemy.Table(
    "department", meta_data,
    slqalchemy.Column("id", slqalchemy.Integer, primary_key=True),
    slqalchemy.Column("name", slqalchemy.String(128), unique=True, nullable=False)
)

employee_table = sqlalchemy.Table(
    "employee", meta_data,
    slqalchemy.Column("id", slqalchemy.Integer, primary_key=True),
    slqalchemy.Column(
        "department_id", sqlalchemy.Integer, 
        sqlalchemy.ForeignKey("department.id"),
        nullable=False
    ),
    slqalchemy.Column("name", slqalchemy.String(128), nullable=False)
)

关联表查询

# 查询 hr 部门的所有员工及部门信息
join = employee_table.join(
    department_table, employee_table.c.department_id == department_table.c.id
)
query = sqlalchemy.select(join).where(department_table.c.name == "hr")
print(conn.execute(query).fetchall())

# 查询 hr 部门的所有员工信息
join = employee_table.join(
    department_table, employee_table.c.department_id == department_table.c.id
)
query = sqlalchemy.select(employee_table).select_from(join).where(department_table.c.name == "hr")
print(conn.execute(query).fetchall())

# 查询员工 Mary 所在的部门信息
join = employee_table.join(
    department_table, employee_table.c.department_id == department_table.c.id
)
query = sqlalchemy.select(department_table).select_from(join).where(employee_table.c.name == "Mary")
print(conn.execute(query).fetchone())

映射类

定义映射类

from sqlalchemy import create_engine, Column, Integer, String, Date
from sqlalchemy.orm import declarative_base, sessionmaker
# from sqlalchemy.ext.declaratvie import declarative_base

engine = create_engine("", echo=True)
Base = declarative_base()

class Person(Base):
    __tablename__ "person"

    id = Column(Integer, primary_key=True)
    name = Column(String(128), unique=True, nullable=False)
    birthday = Column(Date, nullable=False)
    address = Column(String(255), nullable=True)

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)