点击刷新
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
SQLAlchemy是一个独立的orm，对比django的orm框架而言，django无法把它的orm拿出来独立应用在别的web框架下
三层结构，第一层就是orm，第二层就sqlalchemy核心功能维护连接池(connecntio poolling)、Dialect负责跟dbapi打交道，原生sql等。orm再强有的时候也必须使用sql处理复杂查询。第三层dbapi。
SQLAlchemy本身无法操作数据库,本身就是一个翻译官将函数或类翻译成sql语言，其必须以来pymsql(py3)等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作
如(数据库需要自己建立)：
MySQL-Python mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname> pymysql mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] MySQL-Connector mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname> cx_Oracle oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...] 更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html
使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。如果orm满足不了就可以使用这个
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from sqlalchemy import create_engine 4 5 6 engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test", max_overflow=5) 7 8 # 执行SQL 9 # cur = engine.execute( 10 # "INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)" 11 # ) 12 13 # 新插入行自增ID 14 # cur.lastrowid 15 16 # 执行SQL 17 # cur = engine.execute( 18 # "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[('1.1.1.22', 3),('1.1.1.221', 3),] 19 # ) 20 21 22 # 执行SQL 23 # cur = engine.execute( 24 # "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)", 25 # host='1.1.1.99', color_id=3 26 # ) 27 28 # 执行SQL 29 # cur = engine.execute('select * from hosts') 30 # 获取第一行数据 31 # cur.fetchone() 32 # 获取第n行数据 33 # cur.fetchmany(3) 34 # 获取所有数据 35 # cur.fetchall()
# -*- coding:utf-8 -*- import sqlalchemy from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine #查看sqlalchemy 版本信息 print(sqlalchemy.__version__) #echo=True返回执行的过程，不想了解的就可以不用看了 #数据库连接 engine = create_engine("mysql+pymysql://cloud:123456@10.64.8.10:3306/test",echo=True) #生成一个orm基类 Base=declarative_base() # 创建单表 class Users(Base): #表的名字 __tablename__ = 'users' #字段属性Column创建类,autoincrement自增长，主键默认就是自增长 id = Column(Integer, primary_key=True,autoincrement=True)
#唯一，非空 name = Column(String(32),unique=True，nullable=True)
#默认值 password = Column(String(16),default='123456')
#索引
images_id=Column(Integer,Index=True) #查询表时显示的内容，记住这里只是显示返回还是一个对象 def __repr__(self): return self.name
__table_args__ = (
#唯一联合索引 UniqueConstraint('id', 'name', name='uix_id_name'),
#普通联合索引,值1索引名 ，剩下的是字段名字 Index('ix_id_name', 'name', 'password'), )
#创建所有表结构
Base.metadata.create_all(engine)
#删除所有表
#Base.metadata.drop_all(engine)
echo=True效果：
增
1 add_user=Users(name='u1',password=123456) 2 3 mysession=sessionmaker(bind=engine) 4 session=mysession() 5 6 # 增加单条数据 7 # session.add(add_user) 8 # 增加多条数据 9 # session.add_all([ 10 # Users(name='u2',password=123456), 11 # Users(name='u3',password=123456), 12 # Users(name='u4',password=123456), 13 # ]) 14 # 提交保存 15 # session.commit() 16 #打印Users表中的所有数据，如果表中定义没有def __repr__,那么将打印class对象。 17 print(session.query(Users).all()) 18 #循环获取数据,按照Users.id排序 19 for obj in session.query(Users).order_by(Users.id): 20 print(obj.name,obj.password) 21 #结果 22 ('u1', '123456') 23 ('u2', '123456') 24 ('u3', '123456') 25 ('u4', '123456') 26 27 28 #按照字段内容匹配查询 29 for obj in session.query(Users).filter(Users.name.in_(['u1','u2'])): 30 print(obj.name,obj.password) 31 #结果 32 ('u1', '123456') 33 ('u2', '123456') 34 35 #取反，出了u1和u2的数据 36 for obj in session.query(Users).filter(~Users.name.in_(['u1','u2'])): 37 print(obj.name,obj.password) 38 #结果 39 ('u3', '123456') 40 ('u4', '123456') 41 42 #and 和or匹配查询 43 from sqlalchemy import and_, or_ 44 for obj in session.query(Users).filter(and_(Users.name=='u2',Users.password=='123456')): 45 print(obj.name,obj.password) 46 #结果 47 ('u2', '123456')
删
1 session.query(Users).filter(Users.id > 2).delete() 2 session.commit()
改
1 session.query(Users).filter(Users.id > 2).update({"name" : "099"}) 2 session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False) 3 session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate") 4 session.commit()
查
1 ret = session.query(Users).all() 2 ret = session.query(Users.name, Users.extra).all() 3 ret = session.query(Users).filter_by(name='alex').all() 4 ret = session.query(Users).filter_by(name='alex').first() 5 6 循环输出 7 for obj in session.query(Users).order_by(Users.id): 8 print(obj.name,obj.password)
其它
1 # 条件 2 ret = session.query(Users).filter_by(name='alex').all() 3 ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() 4 ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() 5 ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() 6 ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() 7 ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() 8 from sqlalchemy import and_, or_ 9 ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() 10 ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all() 11 ret = session.query(Users).filter( 12 or_( 13 Users.id < 2, 14 and_(Users.name == 'eric', Users.id > 3), 15 Users.extra != "" 16 )).all() 17 18 19 # 通配符 20 ret = session.query(Users).filter(Users.name.like('e%')).all() 21 ret = session.query(Users).filter(~Users.name.like('e%')).all() 22 23 # 限制 24 ret = session.query(Users)[1:2] 25 26 # 排序 27 ret = session.query(Users).order_by(Users.name.desc()).all() 28 ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() 29 30 # 分组 31 from sqlalchemy.sql import func 32 33 ret = session.query(Users).group_by(Users.extra).all() 34 ret = session.query( 35 func.max(Users.id), 36 func.sum(Users.id), 37 func.min(Users.id)).group_by(Users.name).all() 38 39 ret = session.query( 40 func.max(Users.id), 41 func.sum(Users.id), 42 func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all() 43 44 # 连表 45 46 ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all() 47 48 ret = session.query(Person).join(Favor).all() 49 50 ret = session.query(Person).join(Favor, isouter=True).all() 51 52 53 # 组合 54 q1 = session.query(Users.name).filter(Users.id > 2) 55 q2 = session.query(Favor.caption).filter(Favor.nid < 2) 56 ret = q1.union(q2).all() 57 58 q1 = session.query(Users.name).filter(Users.id > 2) 59 q2 = session.query(Favor.caption).filter(Favor.nid < 2) 60 ret = q1.union_all(q2).all()
一对一，多对多表创建
1 # 一对多 2 class Favor(Base): 3 __tablename__ = 'favor' 4 nid = Column(Integer, primary_key=True) 5 caption = Column(String(50), default='red', unique=True) 6 7 8 class Person(Base): 9 __tablename__ = 'person' 10 nid = Column(Integer, primary_key=True) 11 name = Column(String(32), index=True, nullable=True) 12 favor_id = Column(Integer, ForeignKey("favor.nid")) 13 14 15 # 多对多 16 class Group(Base): 17 __tablename__ = 'group' 18 id = Column(Integer, primary_key=True) 19 name = Column(String(64), unique=True, nullable=False) 20 port = Column(Integer, default=22) 21 22 23 class Server(Base): 24 __tablename__ = 'server' 25 26 id = Column(Integer, primary_key=True, autoincrement=True) 27 hostname = Column(String(64), unique=True, nullable=False) 28 29 30 class ServerToGroup(Base): 31 __tablename__ = 'servertogroup' 32 nid = Column(Integer, primary_key=True, autoincrement=True) 33 server_id = Column(Integer, ForeignKey('server.id')) 34 group_id = Column(Integer, ForeignKey('group.id')) 35 36 37 def init_db(): 38 Base.metadata.create_all(engine) 39 40 41 def drop_db(): 42 Base.metadata.drop_all(engine)
© 著作权归作者所有
举报
发表评论
0/200
