SQLAlchemy is the Python SQL toolkit and Object Relational Mapper that gives application developers the full power and flexibility of SQL. SQLAlchemy provides a full suite of well known enterprise-level persistence patterns, designed for efficient and high-performing database access, adapted into a simple and Pythonic domain language.
SQLAlchemy 比Django的orm更接近原生sql的语法书写
1 from sqlalchemy.ext.declarative import declarative_base 2 from sqlalchemy import Column 3 from sqlalchemy import Integer,String,Text,Date,DateTime 4 from sqlalchemy import create_engine 5 7 Base = declarative_base() 8 9 class Users(Base): 10 __tablename__ = ‘users‘ 11 12 id = Column(Integer, primary_key=True) 13 name = Column(String(32), index=True, nullable=False) 14 depart_id = Column(Integer) 15 16 def create_all(): 17 engine = create_engine( 18 "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8", 19 max_overflow=0, # 超过连接池大小外最多创建的连接 20 pool_size=5, # 连接池大小 21 pool_timeout=10, # 池中没有连接最多等待的时间，否则报错 22 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） 23 ) 25 Base.metadata.create_all(engine) 26 27 def drop_all(): 28 engine = create_engine( 29 "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8", 30 max_overflow=0, # 超过连接池大小外最多创建的连接 31 pool_size=5, # 连接池大小 32 pool_timeout=10, # 池中没有连接最多等待的时间，否则报错 33 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） 34 ) 35 Base.metadata.drop_all(engine) 36 37 if __name__ == ‘__main__‘: 38 drop_all() 39 create_all() # 已经有表的话，不会重复创建
注意：SQLAlchemy本身创建表之后，不支持删除表中的字段，再次运行修改表结构的，需要借助第三方模块（靠生成migration文件）
1 from sqlalchemy.orm import sessionmaker 2 from sqlalchemy import create_engine 3 from models import Users 4 5 engine = create_engine( 6 "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8", 7 max_overflow=0, # 超过连接池大小外最多创建的连接 8 pool_size=5, # 连接池大小 9 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 10 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） 11 ) 12 SessionFactory = sessionmaker(bind=engine) 13 14 # 根据Users类对users表进行增删改查 15 session = SessionFactory() 16 17 # ############################## 基本增删改查 ############################### 18 # 1. 增加 19 obj = Users(name=‘alex‘) 20 session.add(obj) 21 session.commit() 22 23 session.add_all([ 24 Users(name=‘小东北‘), 25 Users(name=‘龙泰‘) 26 ]) 27 session.commit() 28 29 # 2. 查 30 result = session.query(Users).all() 31 for row in result: 32 print(row.id,row.name) 33 34 result = session.query(Users).filter(Users.id >= 2) 35 for row in result: 36 print(row.id,row.name) 37 38 result = session.query(Users).filter(Users.id >= 2).first() 39 print(result) 40 41 # 3.删 42 session.query(Users).filter(Users.id >= 2).delete() 43 session.commit() 44 45 # 4.改 46 session.query(Users).filter(Users.id == 4).update({Users.name:‘东北‘}) 47 session.query(Users).filter(Users.id == 4).update({‘name‘:‘小东北‘}) 48 session.query(Users).filter(Users.id == 4).update({‘name‘:Users.name+"DSB"},synchronize_session=False) 49 # synchronize_session 让Users.name+"DSB"变成字符串连接操作 50 51 session.commit() 52 session.close()
1 from sqlalchemy.orm import sessionmaker 2 from sqlalchemy import create_engine 3 from models import Users 4 5 engine = create_engine( 6 "mysql + pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8", 7 max_overflow=0, # 超过连接池大小外最多创建的连接 8 pool_size=5, # 连接池大小 9 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 10 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） 11 ) 12 SessionFactory = sessionmaker(bind=engine) 13 14 # 根据Users类对users表进行增删改查 15 session = SessionFactory() 16 17 18 # 1. 指定列 19 # select id,name as cname from users; 20 result = session.query(Users.id,Users.name.label(‘cname‘)).all() 21 for item in result: 22 print(item[0],item.id,item.cname) 23 24 sql_query = session.query(Users.id,Users.name.label(‘cname‘)) # 查看sql语句 25 26 # 2. 默认条件and 27 session.query(Users).filter(Users.id > 1, Users.name == ‘eric‘).all() 28 # 3. between 29 session.query(Users).filter(Users.id.between(1, 3), Users.name == ‘eric‘).all() 30 # 4. in 31 session.query(Users).filter(Users.id.in_([1,3,4])).all() 32 session.query(Users).filter(~Users.id.in_([1,3,4])).all() 33 # 5. 子查询 34 session.query(Users).filter(Users.id.in_(session.query(Users.id).filter(Users.name==‘eric‘))).all() 35 # 6. and 和 or 36 from sqlalchemy import and_, or_ 37 session.query(Users).filter(Users.id > 3, Users.name == ‘eric‘).all() 38 session.query(Users).filter(and_(Users.id > 3, Users.name == ‘eric‘)).all() 39 session.query(Users).filter(or_(Users.id < 2, Users.name == ‘eric‘)).all() 40 session.query(Users).filter( 41 or_( 42 Users.id < 2, 43 and_(Users.name == ‘eric‘, Users.id > 3), 44 Users.extra != "" 45 )).all() 46 47 # 7. filter_by 48 session.query(Users).filter_by(name=‘ppp‘).all() # 内部还是转换成表达式，调用filter方法 49 50 # 8. 通配符 51 ret = session.query(Users).filter(Users.name.like(‘e%‘)).all() 52 ret = session.query(Users).filter(~Users.name.like(‘e%‘)).all() 53 54 # 9. 切片 55 result = session.query(Users)[1:2] 56 57 # 10.排序 58 ret = session.query(Users).order_by(Users.name.desc()).all() 59 ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() 60 61 # 11. group by 62 from sqlalchemy.sql import func 63 64 ret = session.query( 65 Users.depart_id, 66 func.count(Users.id), # 指定按照什么来聚合，如果不指定则会按照默认底层配置来取，这样可能就不是我们想要的 67 ).group_by(Users.depart_id).all() 68 for item in ret: 69 print(item) 70 71 72 ret = session.query( 73 Users.depart_id, 74 func.count(Users.id), 75 ).group_by(Users.depart_id).having(func.count(Users.id) >= 2).all() # having 76 for item in ret: 77 print(item) 78 79 # 12.union 和 union all 80 """ 81 select id,name from users 82 UNION 83 select id,name from users; 84 """ 85 q1 = session.query(Users.name).filter(Users.id > 2) 86 q2 = session.query(Favor.caption).filter(Favor.nid < 2) 87 ret = q1.union(q2).all() 88 89 q1 = session.query(Users.name).filter(Users.id > 2) 90 q2 = session.query(Favor.caption).filter(Favor.nid < 2) 91 ret = q1.union_all(q2).all() 92 93 94 session.close()
参考：
https://www.cnblogs.com/wupeiqi/articles/8259356.html
原文：https://www.cnblogs.com/carlous/p/10601514.html
