前期准备

安装MySQL，MySQL忘记密码

python可以通过不同的第三方库去连接各种数据库。但程序可能会使用多个数据库，需要做到一次编写适配不同的DBMS。且直接用python调用SQL的写法并不是很好，容易收到注入攻击，所以有必要使用ORM(Object Relational Mapping)来简化代码和使代码便于维护，而python中最著名的ORM为SQLAlchemy，pip install SQLalchemy即可。

前提知识

什么是注入攻击？如python直接调用数据库并插入数据：

1	self.query.execute("insert into major values(\"{}\", \"{}\")".format(major_name, college_name))

但major_name的字符内容为：计算机"self.query.execute("delete table;")"，很容易直接删除数据表或停止软件的运行，造成不可描述的后果。即：肯能执行用户的非法输入，而这在Web开发中更为致命，甚至可通过植入javascript代码操控你的浏览器，达到不可描述的行为。

更详细的数据库攻击、及其避免方式可以参考《Flask Web开发实战入门、进阶与原理解析》这本书（作者是李辉），算是一本不错的技术书籍。

也可以粗暴的看看这篇文章的第一部分，了解为何不能用代码直接操作数据库。

启动数据库

我是安装的便携版的MySQL，所以去对应文件的位置，命令行mysql -uroot -p启动即可：

连接数据库

此部分的代码建议在jupyter-lab这款IDE内编写，因为涉及那种命令行交互的代码较多，就是写一句执行一句，不能从头开始全部执行。

首先导入必要工具库：

1
2
3

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

创建数据库test, 命令为：CREATE DATABASE test;

Engine 是访问数据库的入口，通过它连接test数据库：

1
2
3

# 创建一个连接引擎 
# create_engine("数据库类型+数据库驱动://数据库用户名:数据库密码@IP地址:端口/数据库"，其他参数)
engine = create_engine("mysql+pymysql://root:MyNewPass@localhost:3306/test")

定义数据库表的Schema

如果我们需要一张 users 的表，那么我们可以通过一个 User 类来声明它的元数据。

primary_key 设置为 true 表示为主键
unique 设置为 true 后，表示这个字段不能重复
index 设置为 true ，表示对该字段建立索引
nullable 默认为 true，表示可空
default 设置该字段的默认值

Base = declarative_base()
class User(Base):
    # __tablename__ 指定在 MySQL 中表的名字
    __tablename__ = 'users'
    # 每一个类都需要主键
    id = Column(Integer, primary_key=True)  
    username = Column(String(64), index=True)
    email = Column(String(64), unique=True, index=True)
    location = Column(String(128))
    
    def __repr__(self):
        """定义 repr 让输出更加直观优雅"""
        return '<User {}>'.format(self.username)

在代码中创建用户完毕后，将这个表映射到MySql数据库中：

1	Base.metadata.create_all(bind=engine)

在MySql中查看创建信息，其中 use test;表示使用当前数据库，show tables;表示查看当前数据库的所有数据表。

此外，SQLAlchemy的数据类型和python的数据类型的对应关系如下：

数据类型	python数据类型	说明
Integer	int	整型
String	str	字符串
Float	float	浮点型
Boolean	bool	布尔型
Date	datetime.date	日期
DateTime	datetime.datetime	日期和时间
Time	datetime.time	时间
Text	str	文本类型
LongText	str	长文本类型

增删改查

众所周知，数据库用户的核心功能就是增删改查这四大金刚，而SQLalchemy也很好的提供了支持。查询是其中最重要的功能，所以，关于SQLAlchemy中查询的官方文档如下：https://docs.sqlalchemy.org/en/13/orm/query.html

通常情况下会使用session完成数据库的增删改查

session创建和管理数据库连接的会话
model object 通过session对象访问数据库，并把访问到的数据存入（或修改）提交到数据库中。任意数量的model object被创建，并绑定到session中，session会管理这些对象，而一旦session 里面的objects 有变化，便可以commit/rollback提交或者放弃changs。
rollback 的使用条件是：假如session第一次add了一个对象，第二次又add了同一个对象，此时一定要session.rollback()，直接commit会报错。

一般来说，session在需要访问数据库的时候创建，在session访问数据库的时候，准确来说，应该是“add/update/delete”数据库的时候，会开启database transaction, 假设没有修改autocommit的默认值(False), 那么，database transaction 一直会保持，只有等到session rolled back, committed, or closed（完成状态）的时候才结束。

所以一般建议，当database transaction结束的时候，同时close session, 保证，每次发起请求，都创建一个新的session。

此外需要保证session是一个全局的对象，所以和数据库通信的session在任何时候只有一个，且管理一个session对象远比管理两个对象简单

推荐做法如下，而不是一个类创建一个session：

class ThingOne(object):
    def go(self, session):
        session.query(FooBar).update({"x": 5})

class ThingTwo(object):
    def go(self, session):
        session.query(Widget).update({"q": 18})

def run_my_program():
    session = Session()
    try:
        ThingOne().go(session)
        ThingTwo().go(session)

        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

session线程安全

session 不是线程安全的。在保证session object是全局变量时，在多线程的环境中，默认情况下，多个线程将会共享同一个session，试想一下，假设A线程正在使用session处理数据库，B线程已经执行完成，把session给close了，那么此时A在使用session就会报错。所以必须保证每个线程使用的session都不一样。

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

session_factory = sessionmaker(bind=some_engine)
Session = scoped_session(session_factory)
some_session = Session()
some_other_session = Session()
some_session is some_other_session #True

使用Session()创建的session对象都是一样的，这可以保证代码在多次调用不同的session()依然获得到相同的session 对象。
使用Session()创建的session对象是 Thread-local, session在线程与线程之间没有任何联系。

那么session.remove()即可关闭此次会话。session创建和管理对数据库的连接，当调用close的时候，注意，sqlalchemy不会关闭与mysql的连接，而是把连接返回到连接池。

增

test = User(username='test', email='lanling@gmail.com', location='Beijing/China')
print(test.__dict__)
session = sessionmaker(bind=engine)()
session.add(test)
session.commit()  

muyu = User(username='muyu', email='muyu@gmail.com', location='TangShan/China')
session.add(muyu)
session.commit()

改

print(muyu.username)
muyu.username = 'muyu1'
session.add(muyu)
# 看到修改的字段
session.dirty 
session.commit()
# 输出 muyu1
print(muyu.username) 
# 不合法
print(muyu1.username)

删

1
2
3

# muyu1 不合法
session.delete(muyu) 
session.commit()

查

session 接受到query查询语句, 执行的结果或保持或者关联到session中

首先pip install faker用于生成一些虚假数据。

# 使用 faker 这个模块来生成一些假数据
from faker import Faker 

faker = Faker()

def fake_data_generate(User, session):
    for _ in range(10):
        user = User(username=faker.name(), email=faker.email(), location=faker.address())
        session.add(user)
    session.commit()
fake_data_generate(User, session)

可以在命令行内select *一下，查看是否生成成功。

返回所有查询结果

1	session.query(User).all()

增加查询条件

from sqlalchemy import text
query = session.query(User).filter(text("id > 1"))
# 全部查询结果
query.all()
# 查询结果的第一个
query.first()
# 查询计数
session.query(User).count()
# 查询单个
a = session.query(User).filter(User.id == 1)
print(a.all())
print(a[0].username)

注意事项，text模式的查询，更多的用于打印查询命令。我尝试了在python3.8下，这个输出不了查询结果，但是别人的可以，所以我还是不要用text模式的查询了。

回滚

bob = User(username='test', email='test@gmail.com', location='BeiJing/China')
session.add(bob)
session.rollback()
print (blog in session) # False

这里有更多的查询方式

代码链接

https://github.com/muyuuuu/SQLAlchemy

后续的东西实在太多了，决定分开写了。

本文参考

Access denied for user ‘test’@’localhost’ (using password: YES) except root user
Session参考
 SQLAlchemy参考
 vscode 中一键规范化 python 代码

Just for Life.

数据库系列（一）——SQLAlchemy的使用(CRUD)