from contextlib import contextmanager from sqlalchemy import create_engine, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import relationship, scoped_session from sqlalchemy.orm import sessionmaker def create_session(engine): """封装Session建立过程 此函数只在应用初始化时调用一次便可""" # 建立Session工厂 SessionFactory()就能够建立出session对象, 可是该对象没有实现线程隔离 SessionFactory = sessionmaker(bind=engine) # 生成线程隔离的session对象(每一个线程取本身的session) session = scoped_session(SessionFactory) # ps: 其实返回的session是scoped_session类型对象, 再经过session()才会建立Session对象, 不过scoped_session实现了代理功能, 进行数据操做时, 会自动建立/获取实现了线程隔离的Session对象并执行操做 return session # 建立模型基类 Base = declarative_base() # 建立数据库引擎 # 细节1 sqlalchemy默认实现了线程池功能, 而且能够自动重连(pool_size 线程池中的链接数, max_overflow 超出线程池的额外链接数) # 细节2 若是数据库使用utf-8支持中文, 则设置链接地址时须要标明 ?charset=utf8 不然报错 engine = create_engine('mysql://root:mysql@127.0.0.1:3306/test27?charset=utf8', echo=False, pool_size=5, max_overflow=10) # 建立会话 session = create_session(engine) class User(Base): __tablename__ = 't_user' id = Column(Integer, primary_key=True) name = Column(String(20), unique=True) addresses = relationship('Address') class Address(Base): __tablename__ = 't_address' id = Column(Integer, primary_key=True) detail = Column(String(20)) user_id = Column(Integer, ForeignKey('t_user.id')) @contextmanager # 装饰器形式的上下文管理器 def session_scope(): """使用上下文管理器对session和事务操做进行封装""" try: yield session session.commit() except: session.rollback() raise finally: session.remove() # session工做完成, 销毁session, 释放内存 def index(): """模拟视图函数""" with session_scope() as session: # 获取session对象, 代码块执行完会自动提交 并 销毁session """增长数据""" user1 = User(name='zs') session.add(user1) session.flush() adr1 = Address(detail="中关村3号", user_id=user1.id) adr2 = Address(detail="华强北5号", user_id=user1.id) session.add_all([adr1, adr2]) """查询数据""" # ret = session.query(User, Address).join(Address, User.id == Address.user_id).filter(User.name == 'zs').all() # for user, adr in ret: # print(user.name, adr.detail) """执行原生SQL""" # data = session.execute('select * from t_user') # print(type(data)) # row = data.fetchone() # 取第一条 # print(row.id) # 取主键 # print(row.name) # 取字段 # rows = data.fetchall() # 取全部数据 # for row in rows: # print(row.id, row.name) # print(data.rowcount) # 取条数 if __name__ == '__main__': # 删除全部表 Base.metadata.drop_all(engine) # 建立全部表 Base.metadata.create_all(engine) index()
注释:模型类本身定义,继承生成的基类mysql