基础学习——sqlalchemy入门

「爱情、让人受尽委屈。」 2022-04-05 11:29 156阅读 0赞

SQLAlchemy 教程 —— 基础入门篇

二、ORM 与 SQLAlchemy 简介  
ORM 全称 Object Relational Mapping, 翻译过来叫对象关系映射。简单的说，ORM 将数据库中的表与面向对象语言中的类建立了一种对应关系。这样，我们要操作数据库，数据库中的表或者表中的一条记录就可以直接通过操作类或者类实例来完成。

SQLAlchemy 是Python 社区最知名的 ORM 工具之一，为高效和高性能的数据库访问设计，实现了完整的企业级持久模型。

先安装 SQLAlchemy:

$ sudo apt-get update  
$ sudo pip3 install sqlalchemy \# 安装过程中出现的一些警告不影响接下来的实验。

三、连接与创建

$ sudo vim /etc/mysql/my.cnf  
通过上面的命令打开 MySQL 的配置文件， 添加下面几个配置：

\[client\]  
default-character-set = utf8

\[mysqld\]  
character-set-server = utf8

\[mysql\]  
default-character-set = utf8  
保存退出。现在我们可以启动 MySQL 服务了：

$ sudo service mysql start  
在命令行下输入下面命令启动 MySQL：

$ mysql -uroot -p

看到上面的内容就说明我们的 MySQL 可以正常启动了（注意，上面的密码不需要输入内容，直接回车就行）, 并且我们通过命令：

> create database blog;  
创建一个名为 blog 的数据库为下面的使用作准备，后面的所有操作都是在 blog 中进行的。

另外，我们需要安装一个 Python 与 MySQL 之间的驱动程序：

$ sudo pip3 install pymysql  
3.1 连接数据库  
我们在 /home/shiyanlou/Code 目录下新建 db.py 文件，并向其中写入如下内容：

\# coding: utf-8

from sqlalchemy import create\_engine

engine = create\_engine('mysql+pymysql://root@localhost:3306/blog')

print(engine)  
在上面的程序中，我们连接了默认运行在 3306 端口的 MySQL 中的 blog 数据库。

运行下这个程序，看到下面的信息说明我们已经连接成功了：

3.2 描述表结构  
要使用 ORM, 我们需要将数据表的结构用 ORM 的语言描述出来。SQLAlchmey 提供了一套 Declarative 系统来完成这个任务。我们以创建一个 users 表为例，看看它是怎么用 SQLAlchemy 的语言来描述的：

\# coding: utf-8

from sqlalchemy import create\_engine  
from sqlalchemy.ext.declarative import declarative\_base  
from sqlalchemy import Column, String, Integer

engine = create\_engine('mysql+pymysql://root@localhost:3306/blog?charset=utf8')  
Base = declarative\_base()

class User(Base):

\_\_tablename\_\_ = 'users'

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.username)

Base.metadata.create\_all(engine)  
我们看到，在 User 类中，用 \_\_tablename\_\_ 指定在 MySQL 中表的名字。我们创建了三个基本字段，类中的每一个 Column 代表数据库中的一列，在 Colunm 中，指定该列的一些配置。第一个字段代表类的数据类型，上面我们使用 String, Integer 俩个最常用的类型，其他常用的包括：

Text  
Boolean  
SmallInteger  
DateTime  
nullable=False 代表这一列不可以为空，index=True 表示在该列创建索引。

另外定义 \_\_repr\_\_ 是为了方便调试，你可以不定义，也可以定义的更详细一些。

$ python3 db.py  
运行程序，程序不会有输出信息，但是 sqlalchemy 已经在 MySQL 数据库里面为我们创建了 users 表。

可以进入 MySQL，运行图片中的命令看看表是如何创建的：

四、关系定义  
关系型数据库表一表之间的关系主要分为一对多关系、一对一关系、多对多关系，下面我们将结合普通的博客应用给大家详细讲解下，各个表之间的关系如何定义。

4.1 一对多关系  
对于一个普通的博客应用来说，用户和文章显然是一个一对多的关系，一篇文章属于一个用户，一个用户可以写很多篇文章，那么他们之间的关系可以这样定义：

from sqlalchemy import ForeignKey  
from sqlalchemy.orm import relationship

class User(Base):

\_\_tablename\_\_ = 'users'

id = Column(Integer, primary\_key=True)  
username = Column(String(64), nullable=False, index=True)  
password = Column(String(64), nullable=False)  
email = Column(String(64), nullable=False, index=True)  
articles = relationship('Article')

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.username)

class Article(Base):

\_\_tablename\_\_ = 'articles'

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.title)  
每篇文章有一个外键指向 users 表中的主键 id， 而在 User 中使用 SQLAlchemy 提供的 relationship 描述 关系。而用户与文章的之间的这个关系是双向的，所以我们看到上面的两张表中都定义了 relationship。

SQLAlchemy 提供了 backref 让我们可以只需要定义一个关系：

articles = relationship('Article', backref='author')  
添加了这个就可以不用再在 Article 中定义 relationship 了！

$ wget http://labfile.oss.aliyuncs.com/courses/724/one-to-mul.py \# 获取实验楼提供的完整程序  
然后在 Mysql 的 blog database下执行：

mysql>drop table users; \# 删除数据库中现有的 users 表；  
然后回到命令行，执行：

$ python3 one-to-mul.py \# 执行  
进入mysql 查看 articles 表：

4.2 一对一关系  
在 User 中我们只定义了几个必须的字段， 但通常用户还有很多其他信息，但这些信息可能不是必须填写的，我们可以把它们放到另一张 UserInfo 表中，这样 User 和 UserInfo 就形成了一对一的关系。你可能会奇怪一对一关系为什么不在一对多关系前面？那是因为一对一关系是基于一对多定义的：

class User(Base):

\_\_tablename\_\_ = 'users'

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.username)

class UserInfo(Base):

\_\_tablename\_\_ = 'userinfos'

id = Column(Integer, primary\_key=True)  
name = Column(String(64))  
qq = Column(String(11))  
phone = Column(String(11))  
link = Column(String(64))  
user\_id = Column(Integer, ForeignKey('users.id'))

Base.metadata.create\_all(engine)  
定义方法和一对多相同，只是需要添加 uselist=False 。

和一对多一样，执行前先删除旧的表：

mysql>drop table articles;  
mysql>drop table users;  
然后执行：

$ wget http://labfile.oss.aliyuncs.com/courses/724/one-to-one.py  
$ python3 one-to-one.py  
mysql 中的 userinfos 表：

4.3 多对多关系  
一篇博客通常有一个分类，好几个标签。标签与博客之间就是一个多对多的关系。多对多关系不能直接定义，需要分解成俩个一对多的关系，为此，需要一张额外的表来协助完成，通常对于这种多对多关系的辅助表不会再去创建一个类，而是使用 sqlalchemy 的 Table 类：

\# 在原来代码的基础上导入  
from sqlalchemy import Table

class Article(Base):  
....  
....  
cate\_id = Column(Integer,ForeignKey('categories.id'))  
tags = relationship('Tag',secondary='article\_tag',backref='articles')

class Category(Base):  
\_\_tablename\_\_ = 'categories'

id = Column(Integer,primary\_key=True)  
name = Column(String(64),nullable=False,index=True)  
articles = relationship('Article',backref='category')

def \_\_repr\_\_(self):  
return '%s(%r)'%(self.\_\_class\_\_.\_\_name\_\_,self.name)

article\_tag = Table(  
\# 第一个参数为表名称，第二个参数是 metadata，这俩个是必须的  
'article\_tag', Base.metadata,  
\# 对于辅助表，一般存储要关联的俩个表的 id，并设置为外键  
Column('article\_id', Integer, ForeignKey('articles.id')),  
Column('tag\_id', Integer, ForeignKey('tags.id'))  
)

class Tag(Base):

\_\_tablename\_\_ = 'tags'

id = Column(Integer, primary\_key=True)  
name = Column(String(64), nullable=False, index=True)

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.name)  
删除旧的表：

创建新表：

$ wget http://labfile.oss.aliyuncs.com/courses/724/mul-to-mul.py  
$ python3 mul-to-mul.py  
4.4 映射到数据  
表已经描述好了，在文件末尾使用下面的命令在我们连接的数据库中创建对应的表：

if \_\_name\_\_ == '\_\_main\_\_':  
Base.metadata.create\_all(engine)  
进入 MySQL 看看：

所有的表都已经创建好了！

五、简单 CURD  
当你想打电话给朋友时，你是否得用手机拨通他的号码才能建立起一个会话？同样的，你想和 MySQL 交谈也得先通过 SQLAlchemy 建立一个会话：

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)  
session = Session()  
你可以把 sessionmaker 想象成一个手机，engine 当做 MySQL 的号码，拨通这个“号码”我们就创建了一个 Session 类，下面就可以通过这个类的实例与 MySQL 愉快的交谈了！

5.1 Create  
在 Web 开发过程中，有时候为了查看页面的显示效果或者测试一些功能，一般会模拟一些虚假的数据。Faker 就是用来生成虚假数据的库。 安装它：

$ sudo pip3 install faker  
下面结合 Faker 库创建一些测试数据，在mul-to-mul.py文件末尾添加如下代码：

import random  
\# 导入 faker 工厂对象  
from faker import Factory  
from sqlalchemy.orm import sessionmaker

\# 创建一个 faker 工厂对象  
faker = Factory.create()  
Session = sessionmaker(bind=engine)  
session = Session()

faker\_users = \[User(  
\# 使用 faker 生成一个人名  
username=faker.name(),  
\# 使用 faker 生成一个单词  
password=faker.word(),  
\# 使用 faker 生成一个邮箱  
email=faker.email(),  
) for i in range(10)\]  
\# add\_all 一次性添加多个对象  
session.add\_all(faker\_users)

\# 生成 5 个分类  
faker\_categories = \[Category(name=faker.word()) for i in range(5)\]  
session.add\_all(faker\_categories)

\# 生成 20 个标签  
faker\_tags= \[Tag(name=faker.word()) for i in range(20)\]  
session.add\_all(faker\_tags)

\# 生成 100 篇文章  
for i in range(100):  
article = Article(  
\# sentence() 生成一句话作为标题  
title=faker.sentence(),  
\# 文章内容为随机生成的 10-20句话  
content=' '.join(faker.sentences(nb=random.randint(10, 20))),  
\# 从生成的用户中随机取一个作为作者  
author=random.choice(faker\_users),  
\# 从生成的分类中随机取一个作为分类  
category=random.choice(faker\_categories)  
)  
\# 从生成的标签中随机取 2-5 个作为分类，注意 sample() 函数的用法  
for tag in random.sample(faker\_tags, random.randint(2, 5)):  
article.tags.append(tag)  
session.add(article)

session.commit()  
在上面的代码中我们创建了10个用户，5个分类，20个标签，100篇文章，并且为每篇文章随机选择了2~5个标签。

使用 SQLAlchemy 往数据库中添加数据，我们只需要创建相关类的实例，调用 session.add() 添加一个，或者 session.add\_all() 一次添加多个， 最后 session.commit() 就可以了。

执行如下命令即可生成测试数据：

$ python3 mul-to-mul.py

5.2 Retrieve  
这里为了能够及时看到效果，我们选择使用python3交互式解释器执行，而在日常的开发过程中可以把代码写入文件中运行：

如果我们知道用户 id，就可以用 get 方法， filter\_by 用于按某一个字段过滤，而 filter 可以让我们按多个字段过滤，all 则是获取所有。

获取某一字段值可以直接类的属性获取：

5.3 Update  
更新一个字段：

>>> a = session.query(Article).get(10)  
>>> a.title = 'My test blog post'  
>>> session.add(a)  
>>> session.commit()  
添加一个标签：

>>> a = session.query(Article).get(10)  
>>> a.tags.append(Tag(name='python'))  
>>> session.add(a)  
>>> session.commit()  
5.4 Delete  
>>> a = session.query(Article).get(10)  
>>> session.delete(a)  
>>> session.commit()  
删除直接调用 delete 删除获取到的对象，提交 session 即可。

六、完整代码  
\# coding: utf-8

import random  
from faker import Factory

from sqlalchemy import create\_engine, Table  
from sqlalchemy.ext.declarative import declarative\_base  
from sqlalchemy import ForeignKey  
from sqlalchemy import Column, String, Integer, Text  
from sqlalchemy.orm import sessionmaker, relationship

engine = create\_engine('mysql+pymysql://root@localhost:3306/blog?charset=utf8')  
Base = declarative\_base()

class UserInfo(Base):

\_\_tablename\_\_ = 'userinfos'

class Article(Base):

\_\_tablename\_\_ = 'articles'

id = Column(Integer, primary\_key=True)  
title = Column(String(255), nullable=False, index=True)  
content = Column(Text)  
user\_id = Column(Integer, ForeignKey('users.id'))  
cate\_id = Column(Integer, ForeignKey('categories.id'))  
tags = relationship('Tag', secondary='article\_tag', backref='articles')

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.title)

class User(Base):

\_\_tablename\_\_ = 'users'

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.username)

class Category(Base):

\_\_tablename\_\_ = 'categories'

id = Column(Integer, primary\_key=True)  
name = Column(String(64), nullable=False, index=True)  
articles = relationship('Article', backref='category')

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.name)

article\_tag = Table(  
'article\_tag', Base.metadata,  
Column('article\_id', Integer, ForeignKey('articles.id')),  
Column('tag\_id', Integer, ForeignKey('tags.id'))  
)

class Tag(Base):

\_\_tablename\_\_ = 'tags'

id = Column(Integer, primary\_key=True)  
name = Column(String(64), nullable=False, index=True)

def \_\_repr\_\_(self):  
return '%s(%r)' % (self.\_\_class\_\_.\_\_name\_\_, self.name)

if \_\_name\_\_ == '\_\_main\_\_':  
Base.metadata.create\_all(engine)

faker = Factory.create()  
Session = sessionmaker(bind=engine)  
session = Session()

faker\_users = \[User(  
username=faker.name(),  
password=faker.word(),  
email=faker.email(),  
) for i in range(10)\]  
session.add\_all(faker\_users)

faker\_categories = \[Category(name=faker.word()) for i in range(5)\]  
session.add\_all(faker\_categories)

faker\_tags= \[Tag(name=faker.word()) for i in range(20)\]  
session.add\_all(faker\_tags)

for i in range(100):  
article = Article(  
title=faker.sentence(),  
content=' '.join(faker.sentences(nb=random.randint(10, 20))),  
author=random.choice(faker\_users),  
category=random.choice(faker\_categories)  
)  
for tag in random.sample(faker\_tags, random.randint(2, 5)):  
article.tags.append(tag)  
session.add(article)

session.commit()  
在执行程序前先删除 blog 中所有的表，然后执行:

$ wget http://labfile.oss.aliyuncs.com/courses/724/db.py  
$ python3 db.py