Python3+SQLAlchemy+Sqlite3实现ORM教程

阳光穿透心脏的1/2处 2022-10-30 03:28 291阅读 0赞

一、Sqlite3、SQLAlchemy安装  
Sqlite3是Python3标准库不需要另外安装，只需要安装SQLAlchemy即可。本文sqlalchemy版本为1.2.12

1  
pip install sqlalchemy

二、ORM操作  
除了第一步创建引擎时连接URL不一样，其他操作其他mysql等数据库和sqlite都是差不多的。

2.1 创建数据库连接格式说明  
sqlite创建数据库连接就是创建数据库，而其他mysql等应该是需要数据库已存在才能创建数据库连接；建立数据库连接本文中有时会称为建立数据库引擎。

2.1.1 sqlite创建数据库连接  
以相对路径形式，在当前目录下创建数据库格式如下：

1  
2  
3  
\# sqlite://<nohostname>/<path>  
\# where <path> is relative:  
engine = create\_engine('sqlite:///foo.db')  
以绝对路径形式创建数据库，格式如下：

1  
2  
3  
4  
5  
6  
\#Unix/Mac - 4 initial slashes in total  
engine = create\_engine('sqlite:absolute/path/to/foo.db')  
\#Windows  
engine = create\_engine('sqlite:///C:\\\\path\\\\to\\\\foo.db')  
\#Windows alternative using raw string  
engine = create\_engine(r'sqlite:///C:\\path\\to\\foo.db')  
sqlite可以创建内存数据库（其他数据库不可以），格式如下：

1  
2  
3  
4  
\# format 1  
engine = create\_engine('sqlite://')  
\# format 2  
engine = create\_engine('sqlite:///:memory:', echo=True)  
2.1.2 其他数据库创建数据库连接  
PostgreSQL：

1  
2  
3  
4  
5  
6  
\# default  
engine = create\_engine('postgresql://scott:tiger@localhost/mydatabase')  
\# psycopg2  
engine = create\_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')  
\# pg8000  
engine = create\_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase')  
MySQL：  
1  
2  
3  
4  
5  
6  
7  
8  
\# default  
engine = create\_engine('mysql://scott:tiger@localhost/foo')  
\# mysql-python  
engine = create\_engine('mysql+mysqldb://scott:tiger@localhost/foo')  
\# MySQL-connector-python  
engine = create\_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')  
\# OurSQL  
engine = create\_engine('mysql+oursql://scott:tiger@localhost/foo')  
Oracle:  
1  
2  
engine = create\_engine('oracle://scott:tiger@127.0.0.1:1521/sidname')  
engine = create\_engine('oracle+cx\_oracle://scott:tiger@tnsname')  
MSSQL:  
1  
2  
3  
4  
\# pyodbc  
engine = create\_engine('mssql+pyodbc://scott:tiger@mydsn')  
\# pymssql  
engine = create\_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')  
2.2 创建数据库连接  
我们以在当前目录下创建foo.db为例，后续各步同使用此数据库。

在create\_engine中我们多加了两样东西，一个是echo=Ture，一个是check\_same\_thread=False。

echo=Ture----echo默认为False，表示不打印执行的SQL语句等较详细的执行信息，改为Ture表示让其打印。

check\_same\_thread=False----sqlite默认建立的对象只能让建立该对象的线程使用，而sqlalchemy是多线程的所以我们需要指定check\_same\_thread=False来让建立的对象任意线程都可使用。否则不时就会报错：sqlalchemy.exc.ProgrammingError: (sqlite3.ProgrammingError) SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 35608 and this is thread id 34024. \[SQL: 'SELECT users.id AS users\_id, users.name AS users\_name, users.fullname AS users\_fullname, users.password AS users\_password \\nFROM users \\nWHERE users.name = ?\\n LIMIT ? OFFSET ?'\] \[parameters: \[\{\}\]\] (Background on this error at: http://sqlalche.me/e/f405)

1  
2  
3  
from sqlalchemy import create\_engine  
  
engine = create\_engine('sqlite:///foo.db?check\_same\_thread=False', echo=True)

2.3 定义映射  
先建立基本映射类，后边真正的映射类都要继承它

1  
2  
from sqlalchemy.ext.declarative import declarative\_base  
Base = declarative\_base()

然后创建真正的映射类，我们这里以一下User映射类为例，我们设置它映射到users表。

首先要明确，ORM中一般情况下表是不需要先存在的反而为了类与表对应无误借助通过映射类来创建；当然表已戏存在了也无可以，在下一小结中你可以自己决定如果表存在时要如何操作是重新创建还是使用已有表，但使用已有表你需要确保和类的变量名与表的各字段名要对得上。

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
from sqlalchemy import Column, Integer, String  
  
\# 定义映射类User，其继承上一步创建的Base  
class User(Base):  
\# 指定本类映射到users表  
\_\_tablename\_\_ = 'users'  
\# 如果有多个类指向同一张表，那么在后边的类需要把extend\_existing设为True，表示在已有列基础上进行扩展  
\# 或者换句话说，sqlalchemy允许类是表的字集  
\# \_\_table\_args\_\_ = \{'extend\_existing': True\}  
\# 如果表在同一个数据库服务（datebase）的不同数据库中（schema），可使用schema参数进一步指定数据库  
\# \_\_table\_args\_\_ = \{'schema': 'test\_database'\}  
  
\# 各变量名一定要与表的各字段名一样，因为相同的名字是他们之间的唯一关联关系  
\# 从语法上说，各变量类型和表的类型可以不完全一致，如表字段是String(64)，但我就定义成String(32)  
\# 但为了避免造成不必要的错误，变量的类型和其对应的表的字段的类型还是要相一致  
\# sqlalchemy强制要求必须要有主键字段不然会报错，如果要映射一张已存在且没有主键的表，那么可行的做法是将所有字段都设为primary\_key=True  
\# 不要看随便将一个非主键字段设为primary\_key，然后似乎就没报错就能使用了，sqlalchemy在接收到查询结果后还会自己根据主键进行一次去重  
\# 指定id映射到id字段; id字段为整型，为主键，自动增长（其实整型主键默认就自动增长）  
id = Column(Integer, primary\_key=True, autoincrement=True)  
\# 指定name映射到name字段; name字段为字符串类形，  
name = Column(String(20))  
fullname = Column(String(32))  
password = Column(String(32))  
  
\# \_\_repr\_\_方法用于输出该类的对象被print()时输出的字符串，如果不想写可以不写  
def \_\_repr\_\_(self):  
return "<User(name='%s', fullname='%s', password='%s')>" % (  
self.name, self.fullname, self.password)

在上面的定义我\_\_tablename\_\_属性是写死的，但有时我们可能想通过外部给类传递表名，此时可以通过以下变通的方法来实现：

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
def get\_dynamic\_table\_name\_class(table\_name):  
\# 定义一个内部类  
class TestModel(Base):  
\# 给表名赋值  
\_\_tablename\_\_ = table\_name  
\_\_table\_args\_\_ = \{'extend\_existing': True\}  
  
username = Column(String(32), primary\_key=True)  
password = Column(String(32))  
\# 把动态设置表名的类返回去  
return TestModel  
2.4 创建数据表  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
\# 查看映射对应的表  
User.\_\_table\_\_  
  
\# 创建数据表。一方面通过engine来连接数据库，另一方面根据哪些类继承了Base来决定创建哪些表  
\# checkfirst=True，表示创建表前先检查该表是否存在，如同名表已存在则不再创建。其实默认就是True  
Base.metadata.create\_all(engine, checkfirst=True)  
  
\# 上边的写法会在engine对应的数据库中创建所有继承Base的类对应的表，但很多时候很多只是用来则试的或是其他库的  
\# 此时可以通过tables参数指定方式，指示仅创建哪些表  
\# Base.metadata.create\_all(engine,tables=\[Base.metadata.tables\['users'\]\],checkfirst=True)  
\# 在项目中由于model经常在别的文件定义，没主动加载时上边的写法可能写导致报错，可使用下边这种更明确的写法  
\# User.\_\_table\_\_.create(engine, checkfirst=True)  
  
\# 另外我们说这一步的作用是创建表，当我们已经确定表已经在数据库中存在时，我完可以跳过这一步  
\# 针对已存放有关键数据的表，或大家共用的表，直接不写这创建代码更让人心里踏实

从上边的讨论可以知道，我们可以定义model然后根据model来创建数据表（当然也可以不创建），那可不可以反过来根据已有的表来自动生成model代码呢，答案是可以的，使用sqlacodegen。

sqlacodegen安装操作如下：

1  
2  
3  
4  
5  
6  
7  
\# 如果网络通，直接pip安装  
pip install sqlacodegen  
  
\# 如果网络不通，先在网络通的机器上使用pip下载sqlacodegen及期依赖包  
pip download sqlacodegen  
\# 上传到真正要安装的机器后再用pip安装，依赖包也会自动安装。版本可能会变化改成自己具体的包名  
pip install sqlacodegen-2.1.0-py2.py3-none-any.whl  
sqlacodegen生成model操作如下：

1  
2  
3  
4  
5  
6  
7  
\# linux应该被安装在/usr/local/bin/sqlacodegen  
\# mysql+pymysql示例  
\# 可使用--tables指定要生成model的表，不指定时为所有表都生成model  
\# 可使用--outfile指定代码输出到的文件，不指定时输出到stdout  
\# 注意只有当表有主键时sqlacodegen才生成如下的class，不然会使用旧的生成Table()类实例的形式  
\# 更多说明可使用-h参看  
sqlacodegen mysql+pymysql://user:password@localhost/dbname \[--tables table\_name1,table\_name2\] \[--outfile model.py\]  
如我的一个示例操作如下，成功为指定表生成model：

2.5 建立会话  
增查改删（CRUD）操作需要使用session进行操作

1  
2  
3  
4  
5  
6  
7  
from sqlalchemy.orm import sessionmaker  
  
\# engine是2.2中创建的连接  
Session = sessionmaker(bind=engine)  
  
\# 创建Session类实例  
session = Session()

2.6 增（向users表中插入记录）  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
\# 创建User类实例  
ed\_user = User(name='ed', fullname='Ed Jones', password='edspassword')  
  
\# 将该实例插入到users表  
session.add(ed\_user)  
  
\# 一次插入多条记录形式  
session.add\_all(  
\[User(name='wendy', fullname='Wendy Williams', password='foobar'),  
User(name='mary', fullname='Mary Contrary', password='xxg527'),  
User(name='fred', fullname='Fred Flinstone', password='blah')\]  
)  
  
\# 当前更改只是在session中，需要使用commit确认更改才会写入数据库  
session.commit()

2.7 查（查询users表中的记录）  
2.7.1 查实现  
query将转成select xxx from xxx部分，filter/filter\_by将转成where部分，limit/order by/group by分别对应limit()/order\_by()/group\_by()方法。这句话非常的重要，理解后你将大量减少sql这么写那在sqlalchemy该怎么写的疑惑。

filter\_by相当于where部分，外另可用filter。他们的区别是filter\_by参数写法类似sql形式，filter参数为python形式。

更多匹配写法见：https://docs.sqlalchemy.org/en/13/orm/tutorial.html\#common-filter-operators

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
our\_user = session.query(User).filter\_by(name='ed').first()  
  
our\_user  
  
\# 比较ed\_user与查询到的our\_user是否为同一条记录  
ed\_user is our\_user  
  
\# 只获取指定字段  
\# 但要注意如果只获取部分字段，那么返回的就是元组而不是对象了  
\# session.query(User.name).filter\_by(name='ed').all()  
\# like查询  
\# session.query(User).filter(User.name.like("ed%")).all()  
\# 正则查询  
\# session.query(User).filter(User.name.op("regexp")("^ed")).all()  
\# 统计数量  
\# session.query(User).filter(User.name.like("ed%")).count()  
\# 调用数据库内置函数  
\# 以count()为例，都是直接func.func\_name()这种格式，func\_name与数据库内的写法保持一致  
\# from sqlalchemy import func  
\# session.query(func.count(User3.name)).one()  
\# 字段名为字符串形式  
\# column\_name = "name"  
\# session.query(User).filter(User3.\_\_table\_\_.columns\[column\_name\].like("ed%")).all()  
\# 获取执行的sql语句  
\# 获取记录数的方法有all()/one()/first()等几个方法，如果没加这些方法，得到的只是一个将要执行的sql对象，并没真正提交执行  
\# from sqlalchemy.dialects import mysql  
\# sql\_obj = session.query(User).filter\_by(name='ed')  
\# sql\_command = sql\_obj.statement.compile(dialect=mysql.dialect(), compile\_kwargs=\{"literal\_binds": True\})  
\# sql\_result = sql\_obj.all()

另外要注意该链接Common Filter Operators节中形如equals的query.filter(User.name == 'ed')，在真正使用时都得改成session.query(User).filter(User.name == 'ed')形式，不然只后看到报错“NameError: name 'query' is not defined”。