ORM框架SQLAlchemy

SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

pip install sqlalchemy

简单使用

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() #获取所有行

DB API

SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

#1、MySQL-Python
mysql+mysqldb://:@[:]/

#2、pymysql
mysql+pymysql://:@/[?]

#3、MySQL-Connector
mysql+mysqlconnector://:@[:]/

#4、cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value…]

创建表

四张表:业务线,服务,用户,角色，利用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@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()

增删改查

表结构

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()
print(res) #

#过滤查
res=session.query(Dep).filter(Dep.id > 1,Dep.id <1000) #逗号分隔,默认为and
print([(row.id,row.dname) for row in res])

其他查询相关

准备表和数据

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()

# 准备数据
session.add_all([
 Dep(dname='技术'),
 Dep(dname='销售'),
 Dep(dname='运营'),
 Dep(dname='人事'),
])

session.add_all([
 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()

条件、通配符、limit、排序、分组、连表、组合

一、条件

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) #组合,不去重.union_all(q2) #组合,不去重