SQLAlchemy的基本使用

小咪咪 2022-06-08 11:36 231阅读 0赞

## 1. SQLAlchemy 简介 ##

SQLAlchemy 是 python 语言中一个比较有名的 ORM 框架。使用 SQLAlchemy 则可以分为三种方式：

1.  使用 SQL Expression，通过 SQLAlchemy 提供的方法写sql表达式，间接地写sql；
2.  使用原生SQL，直接书写sql语句；
3.  使用 ORM 对象映射，将类映射到数据库，通过对象来操作数据库。

相对于ORM对象映射来说，前两种方法不需要写类，使用更简单，代码的可读性更好，并且即使不使用ORM，前两种方法也能很好的完成工作，而且复杂的查询语句，依靠单纯的ORM比较难实现。建议把SQLAlchemy当成`DBAPI`使用，SQLAlchemy 内建数据库连接池，解决了连接操作相关繁琐的处理，提供方便的强大的log功能。

本文主要介绍前两种的CURD用法。

## 2. 连接数据库 ##

这里使用的SQLAlchemy版本是v1.1.11

db = create_engine('mysql://admin:admin@10.123.16.160:3306/test?charset=utf8', echo=False)
    conn = db.connect()

通过`create_engine`方法进行数据库连接，返回一个 db 对象，它的两个参数：

> URL格式：`数据库类型://用户名:密码@主机地址:端口/数据库名?charset=编码`  
> echo：是否打印日志

通过`db.connect()`获取到数据库连接对象，后面使用它的`execute()`方法来执行sql语句。

## 3. SQL Expression用法 ##

假设存在一个表`t_email`结构如下：

### 3.1 插入 ###

from sqlalchemy import *
    
    db = create_engine('mysql://admin:admin@10.123.16.160:3306/test?charset=utf8', echo=False)
    conn = db.connect()
    metadata = MetaData(db)
    
    def add():
        table = Table('t_email', metadata, autoload=True)
        i = table.insert()
        conn.execute(i, user=u"李松", email="12345678@qq.com")

最后执行的SQL语句是：`INSERT INTO t_email (user, email) VALUES ('李松', '12345678@qq.com')`

返回最后插入的记录的id：

def add():
        table = Table('t_email', metadata, autoload=True)
        i = table.insert()
        result = conn.execute(i, user="mister", email="mister@qq.com")
        return result.lastrowid

批量插入：

def add():
        table = Table('t_email', metadata, autoload=True)
        i = table.insert()
        data = [
            {
       'user': 'jack', 'email': 'jack@yahoo.com'},
            {
       'user': 'wendy', 'email': 'wendy@163.com'},
        ]
        conn.execute(i, data)

### 3.2 查询 ###

def search():
        table = Table('t_email', metadata, autoload=True)
        s = select([table])
        result = conn.execute(s)
        for row in result:
            print row['user'], row['email']
        result.close()

最后执行的SQL语句是`SELECT t_email.id, t_email.user, t_email.email, t_email.updated_at FROM t_email`，这样把所有字段都查询出来了。

如果只要查询指定的字段：

s = select([table.c.user, table.c.email])

条件查询类似如下：

# ==条件
    s = select([table.c.user, table.c.email]).where(table.c.user == "Jack")
    # in条件
    s = select([table.c.user, table.c.email]).where(table.c.id.in_([1, 2]))
    # between条件
    s = select([table.c.user, table.c.email]).where(table.c.id.between(1, 5))
    # like条件
    s = select([table.c.user, table.c.email]).where(table.c.email.like('%gmail%'))
    # group by条件
    s = select([table.c.user, table.c.email]).group_by(table.c.user)
    # 聚合函数
    s = select([func.count(table.c.id).label('cnt')])

当有多个条件时，可以使用`and_()`、`or_()`、`not_()`进行连接。

多表联合查询：

def search():
        table1 = Table('t_user', metadata, autoload=True)
        table2 = Table('t_email', metadata, autoload=True)
        j_table = table1.join(table2, table1.c.name == table2.c.user)
        s = select([table1.c.name, table1.c.gender, table2.c.email]).select_from(j_table)
        result = conn.execute(s)
        for row in result:
            print row['name'], row['gender'], row['email']
        result.close()

### 3.3 更新 ###

更新操作：

def update():
        table = Table('t_email', metadata, autoload=True)
        u = table.update().values(email='lisong.shine@qq.com').where(table.c.user == u'李松')
        conn.execute(u)

### 3.4 删除 ###

删除操作：

def delete():
        table = Table('t_email', metadata, autoload=True)
        d = table.delete().where(table.c.user == u'李松')
        conn.execute(d)

## 4. 原生SQL ##

遇到复杂的sql语句时，不论是增删改查，都可以直接书写sql语句，然后使用`text()`转换成可执行的对象来执行：

sql = "SELECT * FROM t_email WHERE id=2"
    s = text(sql)
    result = conn.execute(s)
    for row in result:
        print row

这种方法还可以定义变量，然后在execute时赋值：

sql = "SELECT * FROM t_email WHERE id=:id"
    s = text(sql)
    result = conn.execute(s, id=2)
    for row in result:
        print row

## 5. ORM对象映射 ##

使用ORM就是将python类与数据库的table映射，通过对类对象的操作来操作数据库。这种方法免去了写sql语句，但是写类也是个麻烦事，而且复杂的sql语言依靠ORM比较难实现。

### 5.1 插入 ###

通过`session`添加一个新对象即可插入一条新数据

from sqlalchemy import *
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    
    db = create_engine('mysql://root:admin@localhost:3306/test?charset=utf8', echo=True)
    
    Base = declarative_base()
    # 定义类
    class User(Base):
        __tablename__="t_user"
        id = Column(INT, primary_key=True, autoincrement=True)
        name = Column(String(255), nullable=False)
        gender = Column(String(1), nullable=False, default='F')
        age = Column(INT, nullable=False, default=0)
    
    DBSession = sessionmaker(bind=db)
    session = DBSession()
    user = User(name="Selena", gender="M", age=32)
    session.add(user)
    session.commit()
    session.close()

from sqlalchemy import *
    from sqlalchemy.orm import sessionmaker,mapper
    
    db = create_engine('mysql://root:admin@localhost:3306/test?charset=utf8', echo=True)
    metadata = MetaData(db)
    user_table = Table('t_user', metadata, autoload=True)
    
    class User(object):
        pass
    
    # 映射
    mapper(User, user_table)
    DBSession = sessionmaker(bind=db)
    session = DBSession()
    user = User()
    user.name = 'Ellen'
    user.gender = 'M'
    user.age = 38
    session.add(user)
    session.commit()
    session.close()

### 5.2 查询 ###

session = DBSession()
    user = session.query(User).filter(User.id=='5').one()
    user = session.query(User).filter(User.name.like('%ed%'))
    session.close()

想要了解更多可以查看官方文档：[http://docs.sqlalchemy.org/en/rel\_1\_1/index.html][http_docs.sqlalchemy.org_en_rel_1_1_index.html]

[http_docs.sqlalchemy.org_en_rel_1_1_index.html]: http://docs.sqlalchemy.org/en/rel_1_1/index.html