ORM 与 SQLAlchemy 简介:
ORM 叫对象关系映射
,ORM 将对象转换成SQL,然后使用数据API执行SQL并获取执行结果
SQLAlchemy 是Python 社区最知名的 ORM 工具之一,为高效和高性能的数据库访问设计,实现了完整的企业级持久模型。
MySQL-Python mysql+mysqldb://: @ [: ]/ pymysql mysql+pymysql:// : @ / [? ] MySQL-Connector mysql+mysqlconnector:// : @ [: ]/ cx_Oracle oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
#!/usr/bin/env python# -*- coding:utf-8 -*-import sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, INTEGER, Stringfrom sqlalchemy.orm import sessionmakerprint(sqlalchemy.__version__)engin = create_engine('mysql+pymysql://root:123456@127.0.0.1:3306/llt_data',echo=True)#echo=True查看翻译好的sqlBase = declarative_base()#生成一个sqlorm基类#定义一个类(类==表,对象==行)class User(Base): __tablename__ = 'users'#声明表名#列值 id = Column(INTEGER,primary_key=True,autoincrement=True) name = Column(String(10),index=True) fullname = Column(String(10)) password = Column(String(10)) __args__ = ( Index(id, name),#普通索引 uniqueContr(name,fullname)#组合唯一索引 )#可以查看处理的对象具体值 def __repr__(self): return self.nameBase.metadata.create_all(engin)#创建所有表结构# Base.metadata.drop_all(engine)#删除所有表ed_user = User(name='zc',fullname='zc1',password='123')print(ed_user)#触发sessionmake类的__call__方法,return得到Session实例,赋给变量session,就可以进行(session.)等等操作MySession = sessionmaker(bind=engin)session = MySession()session.add(ed_user)#插入数据session.add_all([ User(name='zc1',fullname='zc2',password='123'), User(name='zc1',fullname='zc2',password='123'), User(name='zc1',fullname='zc2',password='123'), User(name='zc1',fullname='zc2',password='123'),])session.commit()#每次必须提交print(session.query(User).all())for row in session.query(User).order_by(User.id): print(row)for row in session.query(User).filter(User.name.in_(['zc','zc1''zc2'])):#全匹配里面的 print(row)for row in session.query(User).filter(~User.name.in_(['zc4','zc1''zc2'])):#取反,不在这三个之中的数据拿出来 print(row)print(session.query(User).filter(User.name == 'zc').count())# 拿到数据,计数#and_, or_必须导入from sqlalchemy import and_, or_for row in session.query(User).filter(and_(User.name == 'zc', User.fullname == 'zc zc2')):#and判断 print(row)for row in session.query(User).filter(or_(User.name == 'zc', User.fullname == 'zc zc2')):#or判断, print(row)
创建表单:
from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column,Integer,String,ForeignKey,UniqueConstraint,Indexfrom sqlalchemy.orm import sessionmaker,relationshipfrom sqlalchemy import create_engineBase=declarative_base()#定义一个类(类==表,对象==行)class UserType(Base): __tablename__='usertype' ##设置表名 id=Column(Integer,primary_key=True,autoincrement=True) ##设置表行 title=Column(String(32),nullable=True,index=True) ##设置表行class Useru(Base): __tablename__='useru' id=Column(Integer,primary_key=True,autoincrement=True) name=Column(String(32),nullable=True,index=True) email=Column(String(16),unique=True) user_type_id=Column(Integer,ForeignKey('usertype.id')) user_type=relationship('UserType',backref='xxoo') ##设置关联语句 relationship找指定的表名(UserType)创建列名(xxoo)实现双向关联,而在自身的表中创建出一列。engine=create_engine("mysql+pymysql://root:123456:@127.0.0.1:3306/lian?charset=utf8",max_overflow=5)##设置连接引擎# 找到所有继承了Base得类,类-SQL语句在数据库创建表Base.metadata.create_all(engine)# 找到所有继承了Base得类,类-SQL语句在数据库删除表# Base.metadata.drop_all(engine)
数据行操作:
# 操作数据行:# # 去引擎中获取一个连接Session=sessionmaker(bind=engine)session=Session()# session.commit()# session.close()########表增加######## obj=Useru(name='话梅',email='92242@qq.com',user_type_id=2) 要插入每行的内容# session.add(obj) 通过对象.add(要插入的变量名)进行添加# session.commit() 通过对象.commit()进行提交#------单个增加-------#======多个增加=======# obj=[# Useru(name='alex1',email='12321@qq.com'),# Useru(name='alex2',email='12322@qq.com'),# Useru(name='alex3',email='12323@qq.com'),# Useru(name='alex4',email='12324@qq.com'),# Useru(name='alex5',email='12325@qq.com')## ]## session.add_all(obj)# session.commit()########查看########## user_list=session.query(Useru)# for row in user_list:# print(row.name,row.id,row.user_type_id)#要进行表的修改和删除首先要查看表的内容########表修改######### session.query(UserType).filter(UserType.id==1).update({ 'title':'黄金用户'})# session.commit()# session.query(UserType).filter(UserType.id==1).update({UserType.title:UserType.title+"x"},synchronize_session=False)# session.commit()# session.query(Useru).filter(Useru.id > 0).update({Useru.name: Useru.name + "x"},synchronize_session=False)# session.query(Useru).filter(Useru.id>2).update({ 'num':Users.num+1},synchronize_session='evaluate') # session.commit()########表删除######## session.query(UserType).filter(UserType.id==1).delete()# session.commit()# type_list=session.query(UserType)# for row in type_list:# print(row.id,row.title)# for j in row.xxoo:# print('-----',j.name)
其他操作 条件: ret=session.query(Useru).filter_by(name='alex').all() #filter_by(键值对方式使用) 可以接收**args类型, .all() ret=session.query(Useru).filter(Useru.id>1,Useru.name=='aelx').all() #filter(条件判断时使用) ret=session.query(Useru).filter(Useru.id.between(1,3),Useru.name=='alex').all() ret=session.query(Useru).filter(Useru.id.in_([1,3,5])).all() #in_就是相当于in ret=session.query(Useru).filter(~Useru.id.in_([1,3,5])).all() #~ 相当于取反的意思 ret=session.query(Useru).filter(Useru.id.in_(session.query(Useru.id).filter_by(name='alex'))).all() #多层套嵌条件查询 from sqlalchemy import and_,or_ ret =session.query(Useru).filter(and_(Useru.id>3,Useru.name=='alex')).all() #and_ 相当于and ret=session.query(Useru).filter(or_(Useru,id<2,Useru.name=='alex')).all() #or_ 相当于or ret=session.query(Useru).filter(or_(Useru.id <2,and_(Useru.name=='alex',Useru.id >3),Useru.extra !='') )#or_括号里都是or的关系,但是在括号里有一个是and 的关系 #通配符: ret=session.query(Useru).filter(Useru.name.like('e%')).all() #查找name以e开头的 ret=session.query(Useru).filter(~Useru.name.like('e%')).all() #查找name不是以e开头的 #限制: ret=session.query(Useru)[1:2] #通过切片的实现分页的功能 #排序: ret=session.query(Useru).order_by(Useru.name.desc()).all() #以降序进行排序 ret=session.query(Useru).order_by(Useru.name.desc(),Useru.id.asc()).all() #开始以降序进行排序,如果过有重复的就按照id以升序排序 #分组:from sqlalchemy.sql import func ret =session.query(Useru).group_by(Useru.name).all() #以名字进行分组 ret=session.query( func.max(Useru.id), func.sum(Useru.id), func.min(Useru.id) ).group_by(Useru.name).all() #以名字进行分组,然后操作合并后的列 ret = session.query( func.max(Useru.id), func.sum(Useru.id), func.min(Useru.id)).group_by(Useru.name).having(func.min(Useru.id) >2).all() #有having二次筛选的 print(ret) #连表: ret=session.query(Useru,UserType).filter(UserType.id==Useru.id).all() #进行连表操作 ret =session.query(Useru).join(UserType).all() #这样是inner join (不显示空) ret = session.query(Useru).join(UserType,isouter=True).all() #这样是left join (显示空)#组合: q1=session.query(Useru.name).filter(Useru.id >2) q2=session.query(UserType.id).filter(UserType.id<1) ret=q1.union(q2).all() #这是去重组合 q1 = session.query(Useru.name).filter(Useru.id > 2) q2 = session.query(UserType.id).filter(UserType.id < 1) ret = q1.union_all(q2).all() #这是不去 #子查询: q3=session.query(Useru.name).filter(Useru.id >2).subquery() restt=session.query(q3).all() print(restt)
创建一对多等:
# 一对多class Favor(Base): __tablename__ = 'favor' nid = Column(Integer, primary_key=True) caption = Column(String(50), default='red', unique=True) class Person(Base): __tablename__ = 'person' nid = Column(Integer, primary_key=True) name = Column(String(32), index=True, nullable=True) favor_id = Column(Integer, ForeignKey("favor.nid")) 一对一给外键后面增加唯一:unique=True, # 多对多class Group(Base): __tablename__ = 'group' id = Column(Integer, primary_key=True) name = Column(String(64), unique=True, nullable=False)# 唯一, 不能为空 port = Column(Integer, default=22)# 端口号=22 class Server(Base): __tablename__ = 'server' id = Column(Integer, primary_key=True, autoincrement=True) hostname = Column(String(64), unique=True, nullable=False) class ServerToGroup(Base): __tablename__ = 'servertogroup' nid = Column(Integer, primary_key=True, autoincrement=True) server_id = Column(Integer, ForeignKey('server.id')) group_id = Column(Integer, ForeignKey('group.id'))
ORM原理:
class User: def __init__(self): self.id=id self.name=name self.email=email def order_by(): passogj=User('1.1.1.1','alex','34434@qq.com')--------------------------------obj.__dict__={ 'id':'1.1.1.1','name':'alex','email':'34434@qq.com'}--------------------------------------------------------------------------ORM的作用就是把类和对象方式解析成SQL语句(不进行连接,连接借助第三方模块)code first 通过手动写SQL语句-------->类db first 通过手动写类--------------->SQL语句
- ORM解决中文编码问题 sqlalchemy 默认使用latin-1进行编码。所以当出现中文时就会报如下错误:
UnicodeEncodeError: 'latin-1' codec can't encode characters in position 39-41: ordinal not in range(256)
解决方法:
在连接数据库的时候直接指定字符编码:
#engine = create_engine("mysql+pymysql://fuzj:123.com@127.0.0.1:3306/fuzj?charset=utf8", max_overflow=5,encoding='utf-8')
ret = session.query(UserType).filter_by(id=1).first()#.first可以取到值print(ret.name)ret = session.query(UserType).filter_by(id=1).all()#.all取到是个列表内的对象,所以无法取到值print(ret.name)ret = session.query(Father.name.label('kkk'),Son.name.label('ppp')).join(Son)#.label设置别名print(ret)
使用__repr__定义返回的数据:
class User(Base): __tablename__ = 'user' nid = Column(Integer,primary_key=True,autoincrement=True) name = Column(String(10),nullable=False) role = Column(Integer,ForeignKey('role.rid')) group = relationship("Role",backref='uuu') #Role为类名 def __repr__(self): output = "(%s,%s,%s)" %(self.nid,self.name,self.role) return output res = session.query(User).all()print(res)输出:[(1,fuzj,1), (2,jie,2), (3,张三,2), (4,李四,1), (5,王五,3)]
ORM 一对多具体使用
mysql表中一对多指的是表A中的数据和表B中的数据存在对应的映射关系,表A中的数据在表B中对应存在多个对应关系,如表A存放用户的角色 DBA,SA,表B中存放用户,表B通过外键关联之表A中,多个用户可以属于同一个角色
-
设计两张表,user表和role表,
user 表中存放用户,role表中存放用户角色,role表中角色对应user表中多个用户,user表中一个用户只对应role表中一个角色,中间通过外键约束
from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String,ForeignKeyfrom sqlalchemy.orm import sessionmaker,relationshipfrom sqlalchemy import create_engineengine = create_engine("mysql+pymysql://fuzj:123.com@127.0.0.1:3306/fuzj?charset=utf8", max_overflow=5,encoding='utf-8') Base = declarative_base()class Role(Base): __tablename__ = 'role' rid = Column(Integer, primary_key=True, autoincrement=True) #主键,自增 role_name = Column(String(10)) def __repr__(self): output = "(%s,%s)" %(self.rid,self.role_name) return outputclass User(Base): __tablename__ = 'user' nid = Column(Integer,primary_key=True,autoincrement=True) name = Column(String(10),nullable=False) role = Column(Integer,ForeignKey('role.rid')) #外键关联 def __repr__(self): output = "(%s,%s,%s)" %(self.nid,self.name,self.role) return outputBase.metadata.create_all(engine)Session = sessionmaker(bind=engine)session = Session()#添加角色数据session.add(Role(role_name='dba'))session.add(Role(role_name='sa'))session.add(Role(role_name='net'))#添加用户数据session.add_all([ User(name='fuzj',role='1'), User(name='jie',role='2'), User(name='张三',role='2'), User(name='李四',role='1'), User(name='王五',role='3'),])session.commit()session.close()
普通连表查询:res = session.query(User,Role).join(Role).all() #查询所有用户,及对应的role idres1 = session.query(User.name,Role.role_name).join(Role).all() #查询所有用户和角色,res2 = session.query(User.name,Role.role_name).join(Role,isouter=True).filter(Role.role_name=='sa').all() #查询所有DBA的用户print(res)print(res1)print(res2)输出结果:[((1,fuzj,1), (1,dba)), ((2,jie,2), (2,sa)), ((3,张三,2), (2,sa)), ((4,李四,1), (1,dba)), ((5,王五,3), (3,net))][('fuzj', 'dba'), ('jie', 'sa'), ('张三', 'sa'), ('李四', 'dba'), ('王五', 'net')][('jie', 'sa'), ('张三', 'sa')]
使用relationship 添加影射关系进行查询
- 首先在User表中添加relationship影射关系
class User(Base): __tablename__ = 'user' nid = Column(Integer,primary_key=True,autoincrement=True) name = Column(String(10),nullable=False) role = Column(Integer,ForeignKey('role.rid')) group = relationship("Role",backref='uuu') #Role为类名
查询:#正向查询print('正向查询')res = session.query(User).all() #查询所有的用户和角色for i in res: print(i.name,i.group.role_name) #此时的i.group 就是role表对应的关系res = session.query(User).filter(User.name=='fuzj').first() #查询fuzj用户和角色print(res.name,res.group.role_name)print('反向查找')#反向查找res = session.query(Role).filter(Role.role_name =='dba').first() #查找dba组下的所有用户print(res.uuu) #此时 print的结果为[(1,fuzj,1), (4,李四,1)]for i in res.uuu: print(i.name,res.role_name)输出结果:正向查询fuzj dbajie sa张三 sa李四 dba王五 netfuzj dba反向查找[(1,fuzj,1), (4,李四,1)]fuzj dba李四 dba
-
-
-
说明
relationship 在user表中创建了新的字段,这个字段只用来存放user表中和role表中的对应关系,在数据库中并不实际存在
正向查找: 先从user表中查到符合name的用户之后,此时结果中已经存在和role表中的对应关系,group对象即role表,所以直接使用obj.group.role_name就可以取出对应的角色反向查找:relationship参数中backref='uuu',会在role表中的每个字段中加入uuu,而uuu对应的就是本字段在user表中对应的所有用户,所以,obj.uuu.name会取出来用户名所谓正向和反向查找是对于relationship关系映射所在的表而说,如果通过该表(user表)去查找对应的关系表(role表),就是正向查找,反正通过对应的关系表(role表)去查找该表(user表)即为反向查找。而relationship往往会和ForeignKey共存在一个表中。
-
-
ORM 多对多具体使用
Mysql多对多关系指的是两张表A和B本没有任何关系,而是通过第三张表C建立关系,通过关系表C,使得表A在表B中存在多个关联数据,表B在表A中同样存在多个关联数据
- 创建三张表 host表 hostuser表 host_to_hostuser表
host表中存放主机,hostuser表中存放主机的用户, host_to_hostuser表中存放主机用户对应的主机,hostuser表中用户对应host表中多个主机,host表中主机对应hostuser表中多个用户,中间关系通过host_to_hostuser表进行关联。host_to_hostuser和host表、user表进行外键约束
from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String,ForeignKeyfrom sqlalchemy.orm import sessionmaker,relationshipfrom sqlalchemy import create_engineclass Host(Base): __tablename__ = 'host' nid = Column(Integer, primary_key=True,autoincrement=True) hostname = Column(String(32)) port = Column(String(32)) ip = Column(String(32))class HostUser(Base): __tablename__ = 'host_user' nid = Column(Integer, primary_key=True,autoincrement=True) username = Column(String(32))class HostToHostUser(Base): __tablename__ = 'host_to_host_user' nid = Column(Integer, primary_key=True,autoincrement=True) host_id = Column(Integer,ForeignKey('host.nid')) host_user_id = Column(Integer,ForeignKey('host_user.nid'))Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)session = Session()#添加数据session.add_all([ Host(hostname='c1',port='22',ip='1.1.1.1'), Host(hostname='c2',port='22',ip='1.1.1.2'), Host(hostname='c3',port='22',ip='1.1.1.3'), Host(hostname='c4',port='22',ip='1.1.1.4'), Host(hostname='c5',port='22',ip='1.1.1.5'),])session.add_all([ HostUser(username='root'), HostUser(username='db'), HostUser(username='nb'), HostUser(username='sb'),])session.add_all([ HostToHostUser(host_id=1,host_user_id=1), HostToHostUser(host_id=1,host_user_id=2), HostToHostUser(host_id=1,host_user_id=3), HostToHostUser(host_id=2,host_user_id=2), HostToHostUser(host_id=2,host_user_id=4), HostToHostUser(host_id=2,host_user_id=3),])session.commit()session.close()
- 普通多次查询
host_id = session.query(Host.nid).filter(Host.hostname=='c2').first() #查找hostbane对应的hostid,返回结果为元组(2,)user_id_list = session.query(HostToHostUser.host_user_id).filter(HostToHostUser.host_id==host_id[0]).all() #查询hostid对应的所有useriduser_id_list = zip(*user_id_list) #user_id_list 初始值为[(2,), (4,), (3,)],使用zip转换为[2,4,3]对象#print(list(user_id_list)) #结果为[(2, 4, 3)]user_list = session.query(HostUser.username).filter(HostUser.nid.in_(list(user_id_list)[0])).all() #查询符合条件的用户print(user_list)#或者:user_id_list = session.query(HostToHostUser.host_user_id).join(Host).filter(Host.hostname=='c2').all()user_id_list = zip(*user_id_list)user_list = session.query(HostUser.username).filter(HostUser.nid.in_(list(user_id_list)[0])).all()print(user_list)
[('db',), ('nb',), ('sb',)]
-
使用relationship映射关系查询
- 首先在关系表Host_to_hostuser中加入relationship关系映射
class HostToHostUser(Base): __tablename__ = 'host_to_host_user' nid = Column(Integer, primary_key=True,autoincrement=True) host_id = Column(Integer,ForeignKey('host.nid')) host_user_id = Column(Integer,ForeignKey('host_user.nid')) host = relationship('Host',backref='h') #对应host表 host_user = relationship('HostUser',backref='u') #对应host_user表
查询:#查找一个服务器上有哪些用户res = session.query(Host).filter(Host.hostname=='c2').first() #返回的是符合条件的服务器对象res2 = res.h #通过relationship反向查找 Host_to_Hostuser中的对应关系for i in res2: #i为host_to_hostuser表和host表中c2主机有对应关系的条目 print(i.host_user.username) #正向查找, 通过relationship ,找到host_to_hostuser中对应的hostuser 即i.host_user#查找此用户有哪些服务器res = session.query(HostUser).filter(HostUser.username=='sb').first()for i in res.u: print(i.host.hostname)
-
-
扩展查询
不查询关系表,直接在hostuser表中指定关系表,然后获取host表
-
在host表中使用 relationship的secondary指定关系表。
class Host(Base): __tablename__ = 'host' nid = Column(Integer, primary_key=True,autoincrement=True) hostname = Column(String(32)) port = Column(String(32)) ip = Column(String(32)) host_user = relationship('HostUser',secondary=lambda :HostToHostUser.__table__,backref='h')注意使用lambda是为了使表的顺序不在闲置查询:host_obj = session.query(Host).filter(Host.hostname=='c1').first()for i in host_obj.host_user: print(i.username)
-
- 首先在关系表Host_to_hostuser中加入relationship关系映射
-
- 首先在User表中添加relationship影射关系
#!/usr/bin/env python# -*- coding:utf-8 -*-from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_engineengine = create_engine('sqlite:///dbyuan674uu.db', echo=True)Base = declarative_base()class Men_to_Wemon(Base): __tablename__ = 'men_to_wemon' nid = Column(Integer, primary_key=True) men_id = Column(Integer, ForeignKey('men.id')) women_id = Column(Integer, ForeignKey('women.id'))class Men(Base): __tablename__ = 'men' id = Column(Integer, primary_key=True) name = Column(String(32)) age= Column(String(16))class Women(Base): __tablename__ ='women' id = Column(Integer, primary_key=True) name = Column(String(32)) age= Column(String(16)) bf=relationship("Men",secondary=Men_to_Wemon.__table__,backref='gf')Base.metadata.create_all(engine)Session = sessionmaker(bind=engine)session = Session()m1=Men(name='alex',age=18)m2=Men(name='wusir',age=18)w1=Women(name='如花',age=40)w2=Women(name='铁锤',age=45)# t1=Men_to_Wemon(men_id=1,women_id=2)m1.gf=[w1,w2]w1.bf=[m1,m2]session.add_all([m1,m2,w1,w2])session.commit()
#!/usr/bin/env python# -*- coding:utf-8 -*-from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_engineengine = create_engine('sqlite:///dbyuan67.db?')Base=declarative_base()class Men(Base): __tablename__ = 'men' id = Column(Integer, primary_key=True) name = Column(String(32)) age= Column(String(16)) # __table_args__ = ( # UniqueConstraint('id', 'name', name='uix_id_name'), # Index('ix_id_name', 'name', 'extra'), # ) def __repr__(self): return self.nameclass Women(Base): __tablename__ ='women' id = Column(Integer, primary_key=True) name = Column(String(32)) age= Column(String(16)) men_id=Column(String(20), ForeignKey('men.id')) # def __repr__(self): # return self.age# Base.metadata.create_all(engine)# Base.metadata.drop_all(engine)Session = sessionmaker(bind=engine)session = Session()# select * from# select id,name from women# sql=session.query(Women).all()# select * from women inner join men on women.men_id = men.idsql = session.query(Women.age,Men.name).join(Men).all()print(sql)# print(sql)# r = session.query(session.query(Women.name.label('t1'), Men.name.label('t2')).join(Men).all()).all()# print(r)# r = session.query(Women).all()# print(r)# m1=Men(name='alex',age=18)# w1=Women(name='如花',age=40)# w2=Women(name='铁锤',age=45)# m1.gf=[Women(name='如花',age=40),Women(name='铁锤',age=45)]# m1=Men(name='alex',age=18)# w1=Women(name='如花',age=40,men_id = 1)# w2=Women(name='铁锤',age=45,men_id = 1)# session.add_all([w1,w2])# session.commit()