SQLAlchemy使用

1、执行原生SQL语句

from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginefrom sqlalchemy.orm import scoped_sessionfrom models import Student,Course,Student2Courseengine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） )SessionFactory = sessionmaker(bind=engine)session = scoped_session(SessionFactory)def task(): """""" # 方式一： """ # 查询 # cursor = session.execute(‘select * from users‘) # result = cursor.fetchall() # 添加 cursor = session.execute(‘INSERT INTO users(name) VALUES(:value)‘, params={"value": ‘wupeiqi‘}) session.commit() print(cursor.lastrowid) """ # 方式二： """ # conn = engine.raw_connection() # cursor = conn.cursor() # cursor.execute( # "select * from t1" # ) # result = cursor.fetchall() # cursor.close() # conn.close() """ # 将连接交还给连接池 session.remove()from threading import Threadfor i in range(20): t = Thread(target=task) t.start()

2、创建数据库表

#!/usr/bin/env python# -*- coding:utf-8 -*-import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Indexfrom sqlalchemy.orm import relationshipBase = declarative_base()# ##################### 单表示例 #########################class Users(Base): __tablename__ = ‘users‘ id = Column(Integer, primary_key=True) name = Column(String(32), index=True) age = Column(Integer, default=18) email = Column(String(32), unique=True) ctime = Column(DateTime, default=datetime.datetime.now) extra = Column(Text, nullable=True) __table_args__ = ( # UniqueConstraint(‘id‘, ‘name‘, name=‘uix_id_name‘), # Index(‘ix_id_name‘, ‘name‘, ‘extra‘), )class Hosts(Base): __tablename__ = ‘hosts‘ id = Column(Integer, primary_key=True) name = Column(String(32), index=True) ctime = Column(DateTime, default=datetime.datetime.now)# ##################### 一对多示例 #########################class Hobby(Base): __tablename__ = ‘hobby‘ id = Column(Integer, primary_key=True) caption = Column(String(50), default=‘篮球‘)class Person(Base): __tablename__ = ‘person‘ nid = Column(Integer, primary_key=True) name = Column(String(32), index=True, nullable=True) hobby_id = Column(Integer, ForeignKey("hobby.id")) # 与生成表结构无关，仅用于查询方便 hobby = relationship("Hobby", backref=‘pers‘)# ##################### 多对多示例 #########################class Server2Group(Base): __tablename__ = ‘server2group‘ id = Column(Integer, primary_key=True, autoincrement=True) server_id = Column(Integer, ForeignKey(‘server.id‘)) group_id = Column(Integer, ForeignKey(‘group.id‘))class Group(Base): __tablename__ = ‘group‘ id = Column(Integer, primary_key=True) name = Column(String(64), unique=True, nullable=False) # 与生成表结构无关，仅用于查询方便 servers = relationship(‘Server‘, secondary=‘server2group‘, backref=‘groups‘)class Server(Base): __tablename__ = ‘server‘ id = Column(Integer, primary_key=True, autoincrement=True) hostname = Column(String(64), unique=True, nullable=False)def init_db(): """ 根据类创建数据库表 :return: """ engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) Base.metadata.create_all(engine)def drop_db(): """ 根据类删除数据库表 :return: """ engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) Base.metadata.drop_all(engine)if __name__ == ‘__main__‘: drop_db() init_db()

方式一：from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginefrom models import Student,Course,Student2Courseengine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） )SessionFactory = sessionmaker(bind=engine)def task(): # 去连接池中获取一个连接 session = SessionFactory() ret = session.query(Student).all() # 将连接交还给连接池 session.close()from threading import Threadfor i in range(20): t = Thread(target=task) t.start()方式二：（推荐，基于Threading.local实现）from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginefrom sqlalchemy.orm import scoped_sessionfrom models import Student,Course,Student2Courseengine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） )SessionFactory = sessionmaker(bind=engine)session = scoped_session(SessionFactory)def task(): ret = session.query(Student).all() # 将连接交还给连接池 session.remove()from threading import Threadfor i in range(20): t = Thread(target=task) t.start()

连接的两种方式

#!/usr/bin/env python# -*- coding:utf-8 -*-from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginefrom sqlalchemy.orm import scoped_sessionfrom models import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)"""# 线程安全，基于本地线程实现每个线程用同一个session# 特殊的：scoped_session中有原来方法的Session中的一下方法：public_methods = ( ‘__contains__‘, ‘__iter__‘, ‘add‘, ‘add_all‘, ‘begin‘, ‘begin_nested‘, ‘close‘, ‘commit‘, ‘connection‘, ‘delete‘, ‘execute‘, ‘expire‘, ‘expire_all‘, ‘expunge‘, ‘expunge_all‘, ‘flush‘, ‘get_bind‘, ‘is_modified‘, ‘bulk_save_objects‘, ‘bulk_insert_mappings‘, ‘bulk_update_mappings‘, ‘merge‘, ‘query‘, ‘refresh‘, ‘rollback‘, ‘scalar‘)"""session = scoped_session(Session)# ############# 执行ORM操作 #############obj1 = Users(name="alex1")session.add(obj1)# 提交事务session.commit()# 关闭sessionsession.close()

基于scoped_session实现线程安全

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom db import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)def task(arg): session = Session() obj1 = Users(name="alex1") session.add(obj1) session.commit()for i in range(10): t = threading.Thread(target=task, args=(i,)) t.start()

多线程执行示例

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import textfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 查询# cursor = session.execute(‘select * from users‘)# result = cursor.fetchall()# 添加cursor = session.execute(‘insert into users(name) values(:value)‘,params={"value":‘wupeiqi‘})session.commit()print(cursor.lastrowid)session.close()

原生sql语句

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import textfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hosts, Hobby, Personengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 添加"""session.add_all([ Hobby(caption=‘乒乓球‘), Hobby(caption=‘羽毛球‘), Person(name=‘张三‘, hobby_id=3), Person(name=‘李四‘, hobby_id=4),])person = Person(name=‘张九‘, hobby=Hobby(caption=‘姑娘‘))session.add(person)hb = Hobby(caption=‘人妖‘)hb.pers = [Person(name=‘文飞‘), Person(name=‘博雅‘)]session.add(hb)session.commit()"""# 使用relationship正向查询"""v = session.query(Person).first()print(v.name)print(v.hobby.caption)"""# 使用relationship反向查询"""v = session.query(Hobby).first()print(v.caption)print(v.pers)"""session.close()

基于relationship操作ForeignKey

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import textfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 添加"""session.add_all([ Server(hostname=‘c1.com‘), Server(hostname=‘c2.com‘), Group(name=‘A组‘), Group(name=‘B组‘),])session.commit()s2g = Server2Group(server_id=1, group_id=1)session.add(s2g)session.commit()gp = Group(name=‘C组‘)gp.servers = [Server(hostname=‘c3.com‘),Server(hostname=‘c4.com‘)]session.add(gp)session.commit()ser = Server(hostname=‘c6.com‘)ser.groups = [Group(name=‘F组‘),Group(name=‘G组‘)]session.add(ser)session.commit()"""# 使用relationship正向查询"""v = session.query(Group).first()print(v.name)print(v.servers)"""# 使用relationship反向查询"""v = session.query(Server).first()print(v.hostname)print(v.groups)"""session.close()

基于relationship操作m2m

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import text, funcfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 关联子查询subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()result = session.query(Group.name, subqry)"""SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid FROM server WHERE server.id = `group`.id) AS anon_1 FROM `group`"""# 原生SQL"""# 查询cursor = session.execute(‘select * from users‘)result = cursor.fetchall()# 添加cursor = session.execute(‘insert into users(name) values(:value)‘,params={"value":‘wupeiqi‘})session.commit()print(cursor.lastrowid)"""session.close()

其他