一、创建表及查询
#!/usr/bin/env python #_*_ coding:utf-8 _*_ #encoding=utf-8 #function: #created by xkq #date: 2018 import sqlalchemy from sqlalchemy import create_engine,and_,or_ from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import sessionmaker engine = create_engine("mysql+pymysql://root:xkq@localhost/orm", encoding='utf-8', echo=True)#数据库名orm要提前创建好 Base = declarative_base() # 生成orm基类 class Host(Base):#创建表hosts结构 __tablename__='hosts'#表名 id=Column(Integer,primary_key=True,autoincrement=True) hostname=Column(String(64),unique=True,nullable=False) ip_address=Column(String(128),unique=True,nullable=False) port=Column(Integer,default=22) Base.metadata.create_all(engine)#创建表hosts if __name__=='__main__': session_class=sessionmaker(bind=engine)#创建于数据库的会话,返回的是一个类 session=session_class()#事例话一个连接 ''' #创建数据 h1=Host(hostname='zabbix',ip_address='192.168.1.1') h2=Host(hostname='mysql',ip_address='192.168.1.2',port='3308') h3=Host(id=3,hostname='mysql2',ip_address='192.168.1.3',port='3308') #session.add(h1)#创建一条数据 session.add_all([h1,h2,h3])#批量创建多条数据 ''' ''' #修改数据 obj=session.query(Host).filter(Host.hostname=='mysql').first()#先找到数据 obj.hostname='mysql_update'#修改数据 ''' ''' #删除数据 obj = session.query(Host).filter(Host.hostname == 'mysql_update').first() session.delete(obj) ''' ''' #查询数据 #obj = session.query(Host).filter(Host.hostname == 'mysql').first()#精确查找 #obj = session.query(Host).filter_by(hostname = 'mysql').first()#精确查找,等价上条语句 #obj=session.query(Host).filter(Host.ip_address.like('%192%')).all()#模糊查找 #obj=session.query(Host).filter(Host.port.in_(['22','3308'])).all()#包含查找 #obj=session.query(Host).filter(~Host.port.in_(['22','3308'])).all()#不包含查找 #obj=session.query(Host).filter(Host.port == None).all()#port 为NULL #obj=session.query(Host).filter(Host.port != None).all()#port NOT NULL #obj=session.query(Host).filter(Host.port.is_(None)).all()#port 为NULL #obj=session.query(Host).filter(Host.port.isnot(None)).all()#port 为 NOT NULL #obj=session.query(Host).filter(and_(Host.hostname.like('za%'),Host.port < 100)).all()#多条件and的查找 #obj=session.query(Host).filter(or_(Host.hostname.like('za%'),Host.port > 100)).all()#多条件or的查找 #obj=session.query(Host).order_by(Host.port).all()#安照port排序 obj=session.query(Host).order_by(Host.port)[1:2]#安照port排序,取地2条数据 print(obj) ''' session.commit()
二、创建表-外键-一对多-jion
#!/usr/bin/env python #_*_ coding:utf-8 _*_ #encoding=utf-8 #function: #created by xkq #date: 2018 import sqlalchemy from sqlalchemy import create_engine,and_,or_,func from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String,ForeignKey from sqlalchemy.orm import sessionmaker,relationship engine = create_engine("mysql+pymysql://root:xkq@localhost/orm", encoding='utf-8', echo=True)#数据库名orm要提前创建好 Base = declarative_base() # 生成orm基类 class Host(Base):#创建表hosts结构 __tablename__='hosts'#表名 id=Column(Integer,primary_key=True,autoincrement=True) hostname=Column(String(64),unique=True,nullable=False) ip_address=Column(String(128),unique=True,nullable=False) port=Column(Integer,default=22) group_id=Column(Integer,ForeignKey('groups.id'))#groups.id作为外键 group=relationship("Group",backref='host')#双向关联,可以通过主机直接找到主机组的名字,如h1.group.name,加上backref='host'可以直接找到一个组下的所有主机 class Group(Base): __tablename__='groups' id=Column(Integer,primary_key=True) name=Column(String(128),unique=True,nullable=False) #host=relationship("Host")#单项关联,可以通过主机组找到一个主机组下的所有主机,如果在Host中用bcakref=‘host'关联,次处则不需要写。 Base.metadata.create_all(engine)#创建表hosts if __name__=='__main__': session_class=sessionmaker(bind=engine)#创建于数据库的会话,返回的是一个类 session=session_class()#事例话一个连接 ''' #往组groups中添加数据 g1=Group(name='group1') g2=Group(name='group2') g3=Group(name='group3') session.add_all([g1,g2,g3]) ''' ''' #往主机hosts中添加数据 h1=Host(hostname='zabbix1',ip_address='192,168.1.1') session.add_all([h1,]) ''' ''' #update,将hosts里的group_id改为1 g1=session.query(Group).filter(Group.id=='group1') h1=session.query(Host).filter(Host.hostname=='zabbix1').update({'group_id':g1.id}) h1=session.query(Host).filter(Host.hostname=='zabbix1').update({'group_id':1})#同上面两行 ''' ''' h1=session.query(Host).filter(Host.hostname=='zabbix1').first() print("主机%s所在组:%s"%(h1.hostname,h1.group.name))#如果没有group=relationship("Group")这一句,则不可实现 g1=session.query(Group).filter(Group.id==1).first() print("主机组%s中的主机有%s个"%(g1.name,len(g1.host))) ''' # jion # obj=session.query(Host).join(Host.group).all()#inner jion obj = session.query(Group.name, func.count(Group.name)).join(Host.group).group_by(Group.name).all() # 查询每个组中的主机数 print("result:", obj) session.commit()
三、创建表-多对多
#!/usr/bin/env python #_*_ coding:utf-8 _*_ #encoding=utf-8 #function: #created by xkq #date: 2018 import sqlalchemy from sqlalchemy import create_engine,and_,or_,func from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String,ForeignKey,Table from sqlalchemy.orm import sessionmaker,relationship engine = create_engine("mysql+pymysql://root:xkq@localhost/orm", encoding='utf-8', echo=True)#数据库名orm要提前创建好 Base = declarative_base() # 生成orm基类 Host_Group=Table( 'host_group',Base.metadata, Column('host_id',Integer,ForeignKey('host.id'),primary_key=True), Column('group_id',Integer,ForeignKey('group.id'),primary_key=True) )#创建中间表实例 class Host(Base):#创建表hosts结构 __tablename__='host'#表名 id=Column(Integer,primary_key=True,autoincrement=True) hostname=Column(String(64),unique=True,nullable=False) ip_address=Column(String(128),unique=True,nullable=False) port=Column(Integer,default=22) groups=relationship("Group",secondary=Host_Group,backref='hosts')#指定中间表事例Host_Group def __repr__(self): return "(id=%s,hostname=%s,ip_address=%s,port=%s)"%(self.id,self.hostname,self.ip_address,self.port) class Group(Base): __tablename__='group' id=Column(Integer,primary_key=True) name=Column(String(128),unique=True,nullable=False) #hosts=relationship("Host",secondary=Host_Group)#单项关联,可以通过主机组找到一个主机组下的所有主机,如果在Host中用bcakref=‘hosts'关联,次处则不需要写。 def __repr__(self):#定义返回的内容 return "(id=%s,name=%s)"%(self.id,self.name) Base.metadata.create_all(engine)#创建表 if __name__=='__main__': session_class=sessionmaker(bind=engine)#创建于数据库的会话,返回的是一个类 session=session_class()#事例话一个连接 ''' #添加组数据 g1=Group(name='group1') g2=Group(name='group2') g3=Group(name='group3') g4=Group(name='group4') session.add_all([g1,g2,g3,g4]) ''' ''' #添加主机数据 h1=Host(hostname='zabbix1',ip_address='192.168.1.1') h2=Host(hostname='zabbix2',ip_address='192.168.1.2',port=53) h3=Host(hostname='zabbix3',ip_address='192.168.1.3',port=161) session.add_all([h1,h2,h3]) ''' ''' #将主机和组进行关联 groups=session.query(Group).all()#是个列表 h1=session.query(Host).filter(Host.hostname=='zabbix1').first() h1.groups=groups#将主机h1添加到所有组中 #h1.groups.pop()#将主机h1从一个组中删除 h2=session.query(Host).filter(Host.hostname=='zabbix2').first() h2.groups=groups[:2]#将主机h2添加到前2个组中 ''' g1=session.query(Group).filter(Group.name=='group2').first() h1=session.query(Host).filter(Host.hostname=='zabbix1').first() print("主机所在的组:",h1.groups)#通过主机查找主机所在的所有组 print("组所包含的主机:",g1.hosts)#通过组查询组所包含的所有主机 # for i in h1.groups: # print(i) session.commit()
注:python3运行sqlalchemy时报waring提示,解决方法:
报错内容: C:Python36libsite - packagespymysqlcursors.py: 170: Warning: ( 1366, "Incorrect string value: '\xD6\xD0\xB9\xFA\xB1\xEA...' for column 'VARIABLE_VALUE' at row 518") result = self._query(query) C:Python36libsite-packagespymysqlcursors.py:170: Warning: (3719, "'utf8' is currently an alias for the character set UTF8MB3, which will be replaced by UTF8MB4 in a future release. Please consider using UTF8MB4 in order to be unambiguous.") result = self._query(query) 解决方法: 打开C:Python36Libsite - packagespymysqlcursors.py将下面两行注释即可 # if not self._defer_warnings: # self._show_warnings()