一、索引
索引，是数据库中专门用于帮助用户快速查询数据的一种数据结构。类似于字典中的目录，查找字典内容时可以根据目录查找到数据的存放位置，然后直接获取即可。
1、索引种类
- 普通索引：仅加速查询
- 唯一索引：加速查询 + 列值唯一（可以有null）
- 主键索引：加速查询 + 列值唯一 + 表中只有一个（不可以有null）
- 组合索引：多列值组成一个索引，专门用于组合搜索，其效率大于索引合并
- 全文索引：对文本的内容进行分词，进行搜索
索引合并，使用多个单列索引组合搜索，组合索引效率 > 索引合并
覆盖索引，select的数据列只用从索引中就能够取得，不必读取数据行，换句话说查询列要被所建的索引覆盖
普通索引
创建表+索引
create table in1(
nid int not null auto_increment primary key,
name varchar(32) not null,
email varchar(64) not null,
extra text,
index ix_name (name)
)
创建表之后，另外创建普通索引
create index index_name on table_name(column_name)
删除普通索引
drop index_name on table_name;
查看索引
show index from table_name;
注意：对于创建索引时如果是BLOB 和 TEXT 类型，必须指定length。
create index ix_extra on in1(extra(32));
唯一索引
创建表+唯一索引
create table in1(
nid int not null auto_increment primary key,
name varchar(32) not null,
email varchar(64) not null,
extra text,
unique ix_name (name)
)
创建表之后，另外创建唯一索引
create unique index 索引名 on 表名(列名)
删除唯一索引
drop unique index 索引名 on 表名
主键索引
创建表+创建主键
create table in1(
nid int not null auto_increment primary key,
name varchar(32) not null,
email varchar(64) not null,
extra text
)
OR
create table in1(
nid int not null auto_increment,
name varchar(32) not null,
email varchar(64) not null,
extra text,
primary key(ni1)
)
创建完表之后，创建主键
alter table 表名 add primary key(列名);
删除主键
alter table 表名 drop primary key;
alter table 表名 modify 列名 int, drop primary key;
组合索引
组合索引是将n个列组合成一个索引
其应用场景为：频繁的同时使用n列来进行查询，如：where n1 = 'alex' and n2 = 666。
-- 创建表 create table in3( nid int not null auto_increment primary key, name varchar(32) not null, email varchar(64) not null, extra text )
-- 创建组合索引 create index ix_name_email on in3(name,email);
如上创建组合索引之后，查询：（最左匹配原则）
- name and email -- 使用索引
- name -- 使用索引
- email -- 不使用索引
注意：对于同时搜索n个条件时，组合索引的性能好于多个单一索引合并。
2、正确使用索引及注意事项
以下情况不会走索引：
- like '%xx' select * from tb1 where name like '%cn'; - 使用函数 select * from tb1 where reverse(name) = 'wupeiqi'; - or select * from tb1 where nid = 1 or email = 'seven@live.com'; 特别的：当or条件中有未建立索引的列才失效，以下会走索引 select * from tb1 where nid = 1 or name = 'seven'; select * from tb1 where nid = 1 or email = 'seven@live.com' and name = 'alex' - 类型不一致 如果列是字符串类型，传入条件是必须用引号引起来，不然... select * from tb1 where name = 999; - != select * from tb1 where name != 'alex' 特别的：如果是主键，则还是会走索引 select * from tb1 where nid != 123 - > select * from tb1 where name > 'alex' 特别的：如果是主键或索引是整数类型，则还是会走索引 select * from tb1 where nid > 123 select * from tb1 where num > 123 - order by select email from tb1 order by name desc; 当根据索引排序时候，选择的映射如果不是索引，则不走索引 特别的：如果对主键排序，则还是走索引： select * from tb1 order by nid desc; - 组合索引最左前缀 如果组合索引为：(name,email) name and email -- 使用索引 name -- 使用索引 email -- 不使用索引
注意事项：
-- 避免使用select * -- count(1)或count(列) 代替 count(*) -- 创建表时尽量时 char 代替 varchar -- 表的字段顺序固定长度的字段优先 -- 组合索引代替多个单列索引（经常使用多个条件查询时） -- 尽量使用短索引 -- 使用连接（JOIN）来代替子查询(Sub-Queries) -- 连表时注意条件类型需一致 -- 索引散列值（重复少）不适合建索引，例：性别不适合
3、limit分页 ？
无论是否有索引，limit分页是一个值得关注的问题
select
*
from
tb1
where
nid < (select nid from (select nid from tb1 where nid < 当前页最小值 order by nid desc limit 每页数据 *【页码-当前页】) A order by A.nid asc limit 1)
order by
nid desc
limit 10;
select
*
from
tb1
where
nid < (select nid from (select nid from tb1 where nid < 970 order by nid desc limit 40) A order by A.nid asc limit 1)
order by
nid desc
limit 10;
上一页：
select
*
from
tb1
where
nid < (select nid from (select nid from tb1 where nid > 当前页最大值 order by nid asc limit 每页数据 *【当前页-页码】) A order by A.nid asc limit 1)
order by
nid desc
limit 10;
select
*
from
tb1
where
nid < (select nid from (select nid from tb1 where nid > 980 order by nid asc limit 20) A order by A.nid desc limit 1)
order by
nid desc
limit 10;
4、执行计划
explain + 查询SQL ------ 用于显示SQL执行信息参数，根据参考信息可以进行SQL优化
mysql> explain select * from tb2; +----+-------------+-------+------+---------------+------+---------+------+------+-------+ | id | select_type | table | type | possible_keys | key | key_len | ref | rows | Extra | +----+-------------+-------+------+---------------+------+---------+------+------+-------+ | 1 | SIMPLE | tb2 | ALL | NULL | NULL | NULL | NULL | 2 | NULL | +----+-------------+-------+------+---------------+------+---------+------+------+-------+ 1 row in set (0.00 sec)
详细：
id 查询顺序标识 如：mysql> explain select * from (select nid,name from tb1 where nid < 10) as B; +----+-------------+------------+-------+---------------+---------+---------+------+------+-------------+ | id | select_type | table | type | possible_keys | key | key_len | ref | rows | Extra | +----+-------------+------------+-------+---------------+---------+---------+------+------+-------------+ | 1 | PRIMARY | <derived2> | ALL | NULL | NULL | NULL | NULL | 9 | NULL | | 2 | DERIVED | tb1 | range | PRIMARY | PRIMARY | 8 | NULL | 9 | Using where | +----+-------------+------------+-------+---------------+---------+---------+------+------+-------------+ 特别的：如果使用union连接其值可能为null select_type 查询类型 SIMPLE 简单查询 PRIMARY 最外层查询 SUBQUERY 映射为子查询 DERIVED 子查询 UNION 联合 UNION RESULT 使用联合的结果 ... table 正在访问的表名 type 查询时的访问方式，性能：all < index < range < index_merge < ref_or_null < ref < eq_ref < system/const ALL 全表扫描，对于数据表从头到尾找一遍 select * from tb1; 特别的：如果有limit限制，则找到之后就不在继续向下扫描 select * from tb1 where email = 'seven@live.com' select * from tb1 where email = 'seven@live.com' limit 1; 虽然上述两个语句都会进行全表扫描，第二句使用了limit，则找到一个后就不再继续扫描。 INDEX 全索引扫描，对索引从头到尾找一遍 select nid from tb1; RANGE 对索引列进行范围查找 select * from tb1 where name < 'alex'; PS: between and in > >= < <= 操作 注意：!= 和 > 符号 INDEX_MERGE 合并索引，使用多个单列索引搜索 select * from tb1 where name = 'alex' or nid in (11,22,33); REF 根据索引查找一个或多个值 select * from tb1 where name = 'seven'; EQ_REF 连接时使用primary key 或 unique类型 select tb2.nid,tb1.name from tb2 left join tb1 on tb2.nid = tb1.nid; CONST 常量 表最多有一个匹配行,因为仅有一行,在这行的列值可被优化器剩余部分认为是常数,const表很快,因为它们只读取一次。 select nid from tb1 where nid = 2 ; SYSTEM 系统 表仅有一行(=系统表)。这是const联接类型的一个特例。 select * from (select nid from tb1 where nid = 1) as A; possible_keys 可能使用的索引 key 真实使用的 key_len MySQL中使用索引字节长度 rows mysql估计为了找到所需的行而要读取的行数 ------ 只是预估值 extra 该列包含MySQL解决查询的详细信息 “Using index” 此值表示mysql将使用覆盖索引，以避免访问表。不要把覆盖索引和index访问类型弄混了。 “Using where” 这意味着mysql服务器将在存储引擎检索行后再进行过滤，许多where条件里涉及索引中的列，当（并且如果）它读取索引时，就能被存储引擎检验，
因此不是所有带where子句的查询都会显示“Using where”。有时“Using where”的出现就是一个暗示：查询可受益于不同的索引。 “Using temporary” 这意味着mysql在对查询结果排序时会使用一个临时表。 “Using filesort” 这意味着mysql会对结果使用一个外部索引排序，而不是按索引次序从表里读取行。mysql有两种文件排序算法，
这两种排序方式都可以在内存或者磁盘上完成，explain不会告诉你mysql将使用哪一种文件排序，也不会告诉你排序会在内存里还是磁盘上完成。 “Range checked for each record(index map: N)” 这个意味着没有好用的索引，新的索引将在联接的每一行上重新估算，N是显示在possible_keys列中索引的位图，并且是冗余的。
慢查询优化的基本步骤：
0.先运行看看是否真的很慢，注意设置SQL_NO_CACHE
1.where条件单表查，锁定最小返回记录表。这句话的意思是把查询语句的where都应用到表中返回的记录数最小的表开始查起，单表每个字段分别查询，看哪个字段的区分度最高
2.explain查看执行计划，是否与1预期一致（从锁定记录较少的表开始查询）
3.order by limit 形式的sql语句让排序的表优先查
4.了解业务方使用场景
5.加索引时参照建索引的几大原则
6.观察结果，不符合预期继续从0分析
5、慢日志查询
a、配置MySQL自动记录慢日志
slow_query_log = OFF 是否开启慢日志记录
long_query_time = 2 时间限制，超过此时间，则记录
slow_query_log_file = /usr/slow.log 日志文件
log_queries_not_using_indexes = OFF 为使用索引的搜索是否记录
查看当前配置信息：show variables like '%query%'
修改当前配置：set global 变量名 = 值
b、查看MySQL慢日志
mysqldumpslow -s at -a /usr/local/var/mysql/MacBook-Pro-3-slow.log
"""
--verbose 版本
--debug 调试
--help 帮助
-v 版本
-d 调试模式
-s ORDER 排序方式
what to sort by (al, at, ar, c, l, r, t), 'at' is default
al: average lock time
ar: average rows sent
at: average query time
c: count
l: lock time
r: rows sent
t: query time
-r 反转顺序，默认文件倒序拍。reverse the sort order (largest last instead of first)
-t NUM 显示前N条just show the top n queries
-a 不要将SQL中数字转换成N，字符串转换成S。don't abstract all numbers to N and strings to 'S'
-n NUM abstract numbers with at least n digits within names
-g PATTERN 正则匹配；grep: only consider stmts that include this string
-h HOSTNAME mysql机器名或者IP；hostname of db server for *-slow.log filename (can be wildcard),
default is '*', i.e. match all
-i NAME name of server instance (if using mysql.server startup script)
-l 总时间中不减去锁定时间；don't subtract lock time from total time
"""
二、流程控制
1、IF条件语句
delimiter \ CREATE PROCEDURE proc_if () BEGIN declare i int default 0; if i = 1 THEN SELECT 1; ELSEIF i = 2 THEN SELECT 2; ELSE SELECT 7; END IF; END\ delimiter ;
2、循环语句
while循环
delimiter \ CREATE PROCEDURE proc_while () BEGIN DECLARE num INT ; SET num = 0 ; WHILE num < 10 DO SELECT num ; SET num = num + 1 ; END WHILE ; END\ delimiter ;
repeat循环
delimiter \ CREATE PROCEDURE proc_repeat () BEGIN DECLARE i INT ; SET i = 0 ; repeat select i; set i = i + 1; until i >= 5 end repeat; END\ delimiter ;
loop循环 ？
BEGIN declare i int default 0; loop_label: loop set i=i+1; if i<8 then iterate loop_label; end if; if i>=10 then leave loop_label; end if; select i; end loop loop_label; END
三、数据备份
1、三种备份方式
- 物理备份：直接复制数据库文件，适用于大型数据库环境，但不能恢复到异构环境中，如windows。
- 逻辑备份：备份建库，建表，插入数据等操作所执行的SQL语句，适用于中小型数据库，效率相对较低。
- 导出表：直接将表导入到文本文件中。
2、用mysqldump实现逻辑备份
#语法： # mysqldump -h 服务器 -u用户名 -p密码 数据库名 > 备份文件.sql #示例： #单库备份 mysqldump -uroot -p123 db1 > db1.sql mysqldump -uroot -p123 db1 table1 table2 > db1-table1-table2.sql #多库备份 mysqldump -uroot -p123 --databases db1 db2 mysql db3 > db1_db2_mysql_db3.sql #备份所有库 mysqldump -uroot -p123 --all-databases > all.sql
恢复逻辑备份
#方法一： [root@egon backup]# mysql -uroot -p123 < /backup/all.sql #方法二： mysql> use db1; mysql> SET SQL_LOG_BIN=0; mysql> source /root/db1.sql #注：如果备份/恢复单个库时，可以修改sql文件 DROP database if exists school; create database school; use school;
3、实现自动化备份
备份计划： 1. 什么时间 2:00 2. 对哪些数据库备份 3. 备份文件放的位置 备份脚本： [root@egon ~]# vim /mysql_back.sql #!/bin/bash back_dir=/backup back_file=`date +%F`_all.sql user=root pass=123 if [ ! -d /backup ];then mkdir -p /backup fi # 备份并截断日志 mysqldump -u${user} -p${pass} --events --all-databases > ${back_dir}/${back_file} mysql -u${user} -p${pass} -e 'flush logs' # 只保留最近一周的备份 cd $back_dir find . -mtime +7 -exec rm -rf {} ;
手动测试： [root@egon ~]# chmod a+x /mysql_back.sql [root@egon ~]# chattr +i /mysql_back.sql [root@egon ~]# /mysql_back.sql 配置cron： [root@egon ~]# crontab -l 2 * * * /mysql_back.sql
4、表的导入和导出
SELECT... INTO OUTFILE 导出文本文件 示例： mysql> SELECT * FROM school.student1 INTO OUTFILE 'student1.txt' FIELDS TERMINATED BY ',' //定义字段分隔符 OPTIONALLY ENCLOSED BY '”' //定义字符串使用什么符号括起来 LINES TERMINATED BY ' ' ; //定义换行符 mysql 命令导出文本文件 示例: # mysql -u root -p123 -e 'select * from student1.school' > /tmp/student1.txt # mysql -u root -p123 --xml -e 'select * from student1.school' > /tmp/student1.xml # mysql -u root -p123 --html -e 'select * from student1.school' > /tmp/student1.html LOAD DATA INFILE 导入文本文件 mysql> DELETE FROM student1; mysql> LOAD DATA INFILE '/tmp/student1.txt' INTO TABLE school.student1 FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '”' LINES TERMINATED BY ' ';
#可能会报错 mysql> select * from db1.emp into outfile 'C:\db1.emp.txt' fields terminated by ',' lines terminated by ' '; ERROR 1238 (HY000): Variable 'secure_file_priv' is a read only variable #数据库最关键的是数据，一旦数据库权限泄露，那么通过上述语句就可以轻松将数据导出到文件中然后下载拿走，因而mysql对此作了限制，只能将文件导出到指定目录 在配置文件中 [mysqld] secure_file_priv='C:\' #只能将数据导出到C:\下 重启mysql 重新执行上述语句
5、数据库迁移
务必保证在相同版本之间迁移 # mysqldump -h 源IP -uroot -p123 --databases db1 | mysql -h 目标IP -uroot -p456
四、PyMySQL模块
1、连接、执行sql、关闭（游标）
import pymysql
user=input('用户名: ').strip() pwd=input('密码: ').strip() #链接 conn=pymysql.connect(host='localhost',user='root',password='123',database='egon',charset='utf8') #游标 cursor=conn.cursor() #执行完毕返回的结果集默认以元组显示 #cursor=conn.cursor(cursor=pymysql.cursors.DictCursor) #执行sql语句 sql='select * from userinfo where name="%s" and password="%s"' %(user,pwd) #注意%s需要加引号 print(sql) res=cursor.execute(sql) #执行sql语句，返回sql查询成功的记录数目 print(res) cursor.close() conn.close() if res: print('登录成功') else: print('登录失败')
2、execute()之sql注入
注意：符号 -- 会注释掉它之后的sql
根本原理：就根据程序的字符串拼接name='%s'，我们输入一个xxx' -- haha,用我们输入的xxx加'在程序中拼接成一个判断条件name='xxx' -- haha'
在一条sql语句中如果遇到select * from t1 where id > 3 -- and name='egon';则--之后的条件被注释掉了 #1、sql注入之用户存在，绕过密码 egon” -- 任意字符 #2、sql注入之用户不存在，绕过用户与密码 xxx” or 1=1 -- 任意字符
解决方法：
原来是我们对sql进行字符串拼接
sql="select * from userinfo where name='%s' and password='%s'" %(user,pwd)
print(sql)
res=cursor.execute(sql)
改写为（execute帮我们做字符串拼接，我们无需且一定不能再为%s加引号了）
sql="select * from userinfo where name=%s and password=%s" #！！！注意%s需要去掉引号，因为pymysql会自动为我们加上
res=cursor.execute(sql,[user,pwd]) #pymysql模块自动帮我们解决sql注入的问题，只要我们按照pymysql的规矩来。
3、增、删、改：conn.commit()
import pymysql #链接 conn=pymysql.connect(host='localhost',user='root',password='123',database='egon') #游标 cursor=conn.cursor() #执行sql语句 #part1 # sql='insert into userinfo(name,password) values("root","123456");' # res=cursor.execute(sql) #执行sql语句，返回sql影响成功的行数 # print(res) #part2 # sql='insert into userinfo(name,password) values(%s,%s);' # res=cursor.execute(sql,("root","123456")) #执行sql语句，返回sql影响成功的行数 # print(res) #part3 sql='insert into userinfo(name,password) values(%s,%s);' res=cursor.executemany(sql,[("root","123456"),("lhf","12356"),("eee","156")]) #执行sql语句，返回sql影响成功的行数 print(res) conn.commit() #提交后才发现表中插入记录成功 cursor.close() conn.close()
4、查：fetchone，fetchmany，fetchall
import pymysql #链接 conn=pymysql.connect(host='localhost',user='root',password='123',database='egon') #游标 cursor=conn.cursor() #执行sql语句 sql='select * from userinfo;' rows=cursor.execute(sql) #执行sql语句，返回sql影响成功的行数rows,将结果放入一个集合，等待被查询 # cursor.scroll(3,mode='absolute') # 相对绝对位置移动 # cursor.scroll(3,mode='relative') # 相对当前位置移动 res1=cursor.fetchone() res2=cursor.fetchone() res3=cursor.fetchone() res4=cursor.fetchmany(2) res5=cursor.fetchall() print(res1) print(res2) print(res3) print(res4) print(res5) print('%s rows in set (0.00 sec)' %rows) conn.commit() #提交后才发现表中插入记录成功 cursor.close() conn.close() ''' (1, 'root', '123456') (2, 'root', '123456') (3, 'root', '123456') ((4, 'root', '123456'), (5, 'root', '123456')) ((6, 'root', '123456'), (7, 'lhf', '12356'), (8, 'eee', '156')) rows in set (0.00 sec) '''
5、获取插入的最后一条数据的自增ID
import pymysql conn=pymysql.connect(host='localhost',user='root',password='123',database='egon') cursor=conn.cursor() sql='insert into userinfo(name,password) values("xxx","123");' rows=cursor.execute(sql) print(cursor.lastrowid) #在插入语句后查看 conn.commit() cursor.close() conn.close()
五、SQLAchemy
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。（程序猿不用写SQL语句哦）
# 安装
pip3 install SQLAlchemy
#1、使用者通过ORM对象提交命令 #2、将命令交给SQLAlchemy Core（Schema/Types SQL Expression Language）转换成SQL #3、使用 Engine/ConnectionPooling/Dialect 进行数据库操作 #3.1、匹配使用者事先配置好的egine #3.2、egine从连接池中取出一个链接 #3.3、基于该链接通过Dialect调用DB API，将SQL转交给它去执行
简化为两个过程：
#第一个阶段（流程1-2）：将SQLAlchemy的对象换成可执行的sql语句 #第二个阶段（流程3）：将sql语句交给数据库执行
如果我们不依赖于SQLAlchemy的转换而自己写好sql语句，那是不是意味着可以直接从第二个阶段开始执行了，事实上正是如此，我们完全可以只用SQLAlchemy执行纯sql语句，如下
from sqlalchemy import create_engine #1 准备 # 需要事先安装好pymysql # 需要事先创建好数据库:create database db1 charset utf8; #2 创建引擎 egine=create_engine('mysql+pymysql://root@127.0.0.1/db1?charset=utf8') #3 执行sql # egine.execute('create table if not EXISTS t1(id int PRIMARY KEY auto_increment,name char(32));') # cur=egine.execute('insert into t1 values(%s,%s);',[(1,"egon1"),(2,"egon2"),(3,"egon3")]) #按位置传值 # cur=egine.execute('insert into t1 values(%(id)s,%(name)s);',name='egon4',id=4) #按关键字传值 #4 新插入行的自增id # print(cur.lastrowid) #5 查询 cur=egine.execute('select * from t1') cur.fetchone() #获取一行 cur.fetchmany(2) #获取多行 cur.fetchall() #获取所有行
SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：
#1、MySQL-Python mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname> #2、pymysql mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] #3、MySQL-Connector mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname> #4、cx_Oracle oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html
1、创建表
# 在ORM框架中： # 类 ==>表 # 对象 ==>表中的一行记录
四张表:业务线, 服务, 用户, 角色，利用ORM创建出它们，并建立好它们之间的关系
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,DateTime,Enum,ForeignKey,UniqueConstraint,ForeignKeyConstraint,Index
from sqlalchemy.orm import sessionmaker
egine=create_engine('mysql+pymysql://root:123@127.0.0.1:3306/db1?charset=utf8',max_overflow=5)
Base=declarative_base()
#创建单表:业务线
class Business(Base):
__tablename__='business'
id=Column(Integer,primary_key=True,autoincrement=True)
bname=Column(String(32),nullable=False,index=True)
#多对一:多个服务可以属于一个业务线,多个业务线不能包含同一个服务
class Service(Base):
__tablename__='service'
id=Column(Integer,primary_key=True,autoincrement=True)
sname=Column(String(32),nullable=False,index=True)
ip=Column(String(15),nullable=False)
port=Column(Integer,nullable=False)
business_id=Column(Integer,ForeignKey('business.id'))
__table_args__=(
UniqueConstraint(ip,port,name='uix_ip_port'),
Index('ix_id_sname',id,sname)
)
#一对一:一种角色只能管理一条业务线,一条业务线只能被一种角色管理
class Role(Base):
__tablename__='role'
id=Column(Integer,primary_key=True,autoincrement=True)
rname=Column(String(32),nullable=False,index=True)
priv=Column(String(64),nullable=False)
business_id=Column(Integer,ForeignKey('business.id'),unique=True)
#多对多:多个用户可以是同一个role,多个role可以包含同一个用户
class Users(Base):
__tablename__='users'
id=Column(Integer,primary_key=True,autoincrement=True)
uname=Column(String(32),nullable=False,index=True)
class Users2Role(Base):
__tablename__='users2role'
id=Column(Integer,primary_key=True,autoincrement=True)
uid=Column(Integer,ForeignKey('users.id'))
rid=Column(Integer,ForeignKey('role.id'))
__table_args__=(
UniqueConstraint(uid,rid,name='uix_uid_rid'),
)
def init_db():
Base.metadata.create_all(egine)
def drop_db():
Base.metadata.drop_all(egine)
if __name__ == '__main__':
init_db()
2、增删改查
表结构： from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmaker egine=create_engine('mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8',max_overflow=5) Base=declarative_base() #多对一:假设多个员工可以属于一个部门,而多个部门不能有同一个员工(只有创建公司才把员工当骆驼用,一个员工身兼数职) class Dep(Base): __tablename__='dep' id=Column(Integer,primary_key=True,autoincrement=True) dname=Column(String(64),nullable=False,index=True) class Emp(Base): __tablename__='emp' id=Column(Integer,primary_key=True,autoincrement=True) ename=Column(String(32),nullable=False,index=True) dep_id=Column(Integer,ForeignKey('dep.id')) def init_db(): Base.metadata.create_all(egine) def drop_db(): Base.metadata.drop_all(egine) drop_db() init_db() Session=sessionmaker(bind=egine) session=Session()
增
#增 row_obj=Dep(dname='销售') #按关键字传参,无需指定id,因其是自增长的 session.add(row_obj) session.add_all([ Dep(dname='技术'), Dep(dname='运营'), Dep(dname='人事'), ]) session.commit()
删
#删 session.query(Dep).filter(Dep.id > 3).delete() session.commit()
改
#改 session.query(Dep).filter(Dep.id > 0).update({'dname':'哇哈哈'}) session.query(Dep).filter(Dep.id > 0).update({'dname':Dep.dname+'_SB'},synchronize_session=False) session.query(Dep).filter(Dep.id > 0).update({'id':Dep.id*100},synchronize_session='evaluate') session.commit()
查
#查所有,取所有字段 res=session.query(Dep).all() #for row in res:print(row.id,row.dname) #查所有,取指定字段 res=session.query(Dep.dname).order_by(Dep.id).all() #for row in res:print(row.dname) res=session.query(Dep.dname).first() #过滤查 res=session.query(Dep).filter(Dep.id > 1,Dep.id <1000) #逗号分隔,默认为and print([(row.id,row.dname) for row in res])
3、条件、通配符、limit、排序、分组、连表、组合
# 准备表结构和数据 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmaker egine=create_engine('mysql+pymysql://root:123@127.0.0.1:3306/db1?charset=utf8',max_overflow=5) Base=declarative_base() #多对一:假设多个员工可以属于一个部门,而多个部门不能有同一个员工(只有创建公司才把员工当骆驼用,一个员工身兼数职) class Dep(Base): __tablename__='dep' id=Column(Integer,primary_key=True,autoincrement=True) dname=Column(String(64),nullable=False,index=True) class Emp(Base): __tablename__='emp' id=Column(Integer,primary_key=True,autoincrement=True) ename=Column(String(32),nullable=False,index=True) dep_id=Column(Integer,ForeignKey('dep.id')) def init_db(): Base.metadata.create_all(egine) def drop_db(): Base.metadata.drop_all(egine) drop_db() init_db() Session=sessionmaker(bind=egine) session=Session() # 准备数据 session.add_all([ Dep(dname='技术'), Dep(dname='销售'), Dep(dname='运营'), Dep(dname='人事'), ]) session.add_all([ Emp(ename='林峰峰',dep_id=1), Emp(ename='李杰姐',dep_id=1), Emp(ename='武琪琪',dep_id=1), Emp(ename='元日天',dep_id=2), Emp(ename='李钢弹',dep_id=3), Emp(ename='张二丫',dep_id=4), Emp(ename='李坦克',dep_id=2), Emp(ename='王大炮',dep_id=4), Emp(ename='牛榴弹',dep_id=3) ]) session.commit()
#一、条件 sql=session.query(Emp).filter_by(ename='林海峰') #filter_by只能传参数:什么等于什么 res=sql.all() #sql语句的执行结果 res=session.query(Emp).filter(Emp.id>0,Emp.ename == '林海峰').all() #filter内传的是表达式,逗号分隔,默认为and, res=session.query(Emp).filter(Emp.id.between(1,3),Emp.ename == '林海峰').all() res=session.query(Emp).filter(Emp.id.in_([1,3,99,101]),Emp.ename == '林海峰').all() res=session.query(Emp).filter(~Emp.id.in_([1,3,99,101]),Emp.ename == '林海峰') #~代表取反,转换成sql就是关键字not from sqlalchemy import and_,or_ res=session.query(Emp).filter(and_(Emp.id > 0,Emp.ename=='林海峰')).all() res=session.query(Emp).filter(or_(Emp.id < 2,Emp.ename=='功夫熊猫')).all() res=session.query(Emp).filter( or_( Emp.dep_id == 3, and_(Emp.id > 1,Emp.ename=='功夫熊猫'), Emp.ename != '' ) ).all() #二、通配符 res=session.query(Emp).filter(Emp.ename.like('%海_%')).all() res=session.query(Emp).filter(~Emp.ename.like('%海_%')).all() #三、limit res=session.query(Emp)[0:5:2] #四、排序 res=session.query(Emp).order_by(Emp.dep_id.desc()).all() res=session.query(Emp).order_by(Emp.dep_id.desc(),Emp.id.asc()).all() #五、分组 from sqlalchemy.sql import func res=session.query(Emp.dep_id).group_by(Emp.dep_id).all() res=session.query( func.max(Emp.dep_id), func.min(Emp.dep_id), func.sum(Emp.dep_id), func.avg(Emp.dep_id), func.count(Emp.dep_id), ).group_by(Emp.dep_id).all() res=session.query( Emp.dep_id, func.count(1), ).group_by(Emp.dep_id).having(func.count(1) > 2).all() #六、连表 #笛卡尔积 res=session.query(Emp,Dep).all() #select * from emp,dep; #where条件 res=session.query(Emp,Dep).filter(Emp.dep_id==Dep.id).all() # for row in res: # emp_tb=row[0] # dep_tb=row[1] # print(emp_tb.id,emp_tb.ename,dep_tb.id,dep_tb.dname) #内连接 res=session.query(Emp).join(Dep) #join默认为内连接,SQLAlchemy会自动帮我们通过foreign key字段去找关联关系 #但是上述查询的结果均为Emp表的字段,这样链表还有毛线意义,于是我们修改为 res=session.query(Emp.id,Emp.ename,Emp.dep_id,Dep.dname).join(Dep).all() #左连接:isouter=True res=session.query(Emp.id,Emp.ename,Emp.dep_id,Dep.dname).join(Dep,isouter=True).all() #右连接:同左连接,只是把两个表的位置换一下 #七、组合 q1=session.query(Emp.id,Emp.ename).filter(Emp.id > 0,Emp.id < 5) q2=session.query(Emp.id,Emp.ename).filter( or_( Emp.ename.like('%海%'), Emp.ename.like('%昊%'), ) ) res1=q1.union(q2) #组合+去重 res2=q1.union_all(q2) #组合,不去重 print([i.ename for i in q1.all()]) #['林海峰', '李杰', '武配齐', '元昊'] print([i.ename for i in q2.all()]) #['林海峰', '元昊'] print([i.ename for i in res1.all()]) #['林海峰', '李杰', '武配齐', '元昊'] print([i.ename for i in res2.all()]) #['林海峰', '李杰', '武配齐', '元昊', '元昊', '林海峰']
4、子查询
形式一: 子查询当做一张表来用,调用subquery()
#示例:查出id大于2的员工,当做子查询的表使用
#原生SQL:
# select * from (select * from emp where id > 2);
#ORM:
res=session.query(
session.query(Emp).filter(Emp.id > 8).subquery()
).all()
形式二:子查询当做in的范围用,调用in_
#示例:#查出销售部门的员工姓名
#原生SQL:
# select ename from emp where dep_id in (select id from dep where dname='销售');
#ORM:
res=session.query(Emp.ename).filter(Emp.dep_id.in_(
session.query(Dep.id).filter_by(dname='销售'), #传的是参数
# session.query(Dep.id).filter(Dep.dname=='销售') #传的是表达式
)).all()
形式三:子查询当做select后的字段,调用as_scalar()
#示例:查询所有的员工姓名与部门名
#原生SQL:
# select ename as 员工姓名,(select dname from dep where id = emp.dep_id) as 部门名 from emp;
#ORM:
sub_sql=session.query(Dep.dname).filter(Dep.id==Emp.dep_id) #SELECT dep.dname FROM dep, emp WHERE dep.id = emp.dep_id
sub_sql.as_scalar() #as_scalar的功能就是把上面的sub_sql加上了括号
res=session.query(Emp.ename,sub_sql.as_scalar()).all()
有三种形式的子查询，注意：子查询的sql必须用括号包起来,尤其在形式三中需要注意这一点
5、正查、反查
# 修改一下表 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import sessionmaker,relationship egine=create_engine('mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8',max_overflow=5) Base=declarative_base() class Dep(Base): __tablename__='dep' id=Column(Integer,primary_key=True,autoincrement=True) dname=Column(String(64),nullable=False,index=True) class Emp(Base): __tablename__='emp' id=Column(Integer,primary_key=True,autoincrement=True) ename=Column(String(32),nullable=False,index=True) dep_id=Column(Integer,ForeignKey('dep.id')) #在ForeignKey所在的类内添加relationship的字段,注意: #1:Dep是类名 #2:depart字段不会再数据库表中生成字段 #3:depart用于Emp表查询Dep表(正向查询),而xxoo用于Dep表查询Emp表(反向查询), depart=relationship('Dep',backref='xxoo') def init_db(): Base.metadata.create_all(egine) def drop_db(): Base.metadata.drop_all(egine) drop_db() init_db() Session=sessionmaker(bind=egine) session=Session() # 准备数据 session.add_all([ Dep(dname='技术'), Dep(dname='销售'), Dep(dname='运营'), Dep(dname='人事'), ]) session.add_all([ Emp(ename='林海峰',dep_id=1), Emp(ename='李杰',dep_id=1), Emp(ename='武配齐',dep_id=1), Emp(ename='元昊',dep_id=2), Emp(ename='李钢弹',dep_id=3), Emp(ename='张二丫',dep_id=4), Emp(ename='李坦克',dep_id=2), Emp(ename='王大炮',dep_id=4), Emp(ename='牛榴弹',dep_id=3) ]) session.commit()
标准连表查询
# 示例:查询员工名与其部门名 res=session.query(Emp.ename,Dep.dname).join(Dep) #迭代器 for row in res: print(row[0],row[1]) #等同于print(row.ename,row.dname)
基于relationship的正查、反查
#SQLAlchemy的relationship在内部帮我们做好表的链接 #查询员工名与其部门名(正向查) res=session.query(Emp) for row in res: print(row.ename,row.id,row.depart.dname) #查询部门名以及该部门下的员工(反向查) res=session.query(Dep) for row in res: # print(row.dname,row.xxoo) print(row.dname,[r.ename for r in row.xxoo])