一、创建表及查询
#!/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()